From 5f50361eef385bef9505d53ed776d348c13b68b2 Mon Sep 17 00:00:00 2001 From: Polichinl Date: Fri, 14 Aug 2026 23:23:24 +0200 Subject: [PATCH 1/3] =?UTF-8?q?perf(wire):=20#269=20=E2=80=94=20assembling?= =?UTF-8?q?=20a=20target=20held=20three=20copies=20of=20it,=20not=20one?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Measured before changing anything, and the issue's diagnosis was a third of the story. WHAT THE MEASUREMENT SHOWED. A synthetic run through the real TargetLease.load — producer-format .tap.zip shards, real read_shard, tracemalloc and peak RSS agreeing within 2% — at 36 shards x 20,000 cells x 200 samples: peak, tracemalloc 3.06x the delivered frame -> 1.13x peak, RSS delta 3.02x -> 1.06x The 3x was three roughly equal thirds, and #269 named only the first: every shard's bytes resident together; every decoded per-shard frame held for the stack; and the np.concatenate result allocated while those were still alive. The ratio held at both 12 and 36 shards, so it is the shape and not the scale. Fixing only the part the issue described would have reached about 2x. Two false starts worth recording, because both would have produced a confident wrong number. My first harness had the fake store return pre-built objects, so the download allocation fell outside the measurement window and the peak read 2.04x. The obvious fix — return bytes(blob) — allocates nothing, because bytes(b) is b when b is already bytes; so does b"" + b. It took a bytearray to get a genuine copy. The 3.06x only appeared once the store actually allocated, which is what a real download does. THE FIX IS ONE LOOP. frames_for_target takes fetch_shard_bytes(name) instead of a filled dict, drops each shard's bytes the moment they are decoded, and writes each shard into a manifest-sized buffer instead of stacking and concatenating. Peak is now the finished frame plus about one shard; the residual 0.13x is 2/n_shards, which is why 12 shards measures 1.36x and 36 measures 1.13x. The buffer's slots are sized from expected_cell_count, which this function already enforced per shard — and the enforcement runs BEFORE anything is written, so a shard whose row count disagrees is refused rather than straddling two months' slots. NOTHING ABOUT THE PRODUCT CHANGED. The wire is byte-frozen (ADR-013) and tests/test_wire_fixture.py compares delivered artifacts against checked-in bytes. It passes unchanged, which is the whole claim: an assembly change with no output change. AT PRODUCTION SCALE, 64,742 cells x 36 months at ADR-013 §5's "~1000 samples per cell" (the one input taken from the contract rather than measured), one target's frame is 8.68 GB, so peak falls from about 26.6 GB to 9.8 GB per target. THE MANAGER'S HISTORICAL FRAME IS NOT THE ELEPHANT. #269 also notes _historical_frame is held from _read through _save. By its declared dimensions — 64,742 cells x 438 months = 28,356,996 rows — that is about 108 MB at one float32 column, 1.2% of a single forecast target frame. Recorded in C-101 rather than filed as its own issue: a separate issue implying comparable cost would misdirect whoever picked it up. GUARDED by asserting the fetch/decode interleaving rather than a byte count — a memory threshold in a test is a flake on a busy machine, while "fetch, decode, fetch, decode" is exactly the property that bounds the peak. Mutation-proven: restoring the up-front dict yields ['fetch','fetch','fetch','decode','decode','decode'] and it fails. The register guard also caught me inventing a cross-reference to a register entry C-126 that does not exist; the 126 is an issue number. Suite 450 passed / 1 skipped / 39 xfailed, ruff clean. Closes #269. Co-Authored-By: Claude Opus 5 (1M context) --- reports/technical_risk_register.md | 45 ++++++++++- tests/test_track_a_source.py | 54 ++++++++++++-- .../contract/track_a_source.py | 74 ++++++++++++++----- .../contract/wire/source_selection.py | 18 +++-- 4 files changed, 158 insertions(+), 33 deletions(-) diff --git a/reports/technical_risk_register.md b/reports/technical_risk_register.md index 6a160ca..5fb205b 100644 --- a/reports/technical_risk_register.md +++ b/reports/technical_risk_register.md @@ -5,9 +5,9 @@ | Project | views-postprocessing | | Owner | Dylan Pinheiro / PRIO MD&D Team | | Last Updated | 2026-08-14 | -| Total Concerns | 100 | +| Total Concerns | 101 | | Open Concerns | 22 | -| Resolved Concerns | 78 | +| Resolved Concerns | 79 | --- @@ -1117,6 +1117,47 @@ See also C-40 (the inheritance/representation coupling this migration unwinds), ## Resolved Concerns +### C-101: Assembling a target held three copies of it — measured, then bounded — RESOLVED + +| Field | Value | +|-------|-------| +| ID | C-101 | +| Tier | 2 — no incorrect output, but the forecast leg needed roughly three times the memory its product occupies, and the failure mode is an OOM kill mid-delivery rather than a refusal. | +| Source | views-postprocessing#269, filed from the views-crafdapi seat 2026-08-14 after the first `un_crafd` delivery attempt | +| Trigger | *(closed)* Any run large enough that three copies of one target's frame did not fit — which on 2026-08-13 meant a machine with 15 GB already in use. | +| Location | `views_postprocessing/contract/track_a_source.py` (`frames_for_target`); `views_postprocessing/contract/wire/source_selection.py` (`TargetLease.load`) | + +**Measured before anything was changed**, because the issue's own diagnosis named one cause and there turned out to be three. A synthetic run through the real `TargetLease.load` — producer-format `.tap.zip` shards, real `read_shard`, `tracemalloc` and peak RSS agreeing to within 2% — at 36 shards x 20,000 cells x 200 samples: + +| | before | after | +|---|---|---| +| peak, tracemalloc | **3.06x** the delivered frame | **1.13x** | +| peak, RSS delta | 3.02x | 1.06x | + +The 3x was **three roughly equal thirds**, and the issue named only the first: + +1. every shard's bytes, resident together — the dict comprehension completed before the first shard was decoded; +2. every decoded per-shard frame, held for the stack; +3. the `np.concatenate` result, allocated while (2) was still alive. + +The ratio held at 12 and 36 shards, so it is the shape and not the scale. Fixing only (1), as the issue proposed, would have taken 3.06x to about 2x. + +**The fix is one loop.** `frames_for_target` now takes `fetch_shard_bytes(name)` instead of a filled dict, drops each shard's bytes the moment they are decoded, and writes each shard into a manifest-sized buffer instead of stacking and concatenating. Peak is now the finished frame plus about one shard — the residual 0.13x is `2/n_shards`, which is why the 12-shard case measures 1.36x and the 36-shard case 1.13x. + +*What makes the buffer safe.* Its slots are sized from `expected_cell_count`, a declaration this function already enforced per shard, and the enforcement runs **before** anything is written — so a shard whose row count disagrees is refused rather than straddling two months' slots. + +*What proves the product did not change.* The wire is byte-frozen (ADR-013) and the golden fixture guards in `tests/test_wire_fixture.py` compare delivered artifacts against checked-in bytes. They pass unchanged, which is the claim: this is an assembly change with no output change. + +**At production scale.** 64,742 cells x 36 months, at ADR-013 §5's *"~1000 samples per cell"* — the sample count is the one input here taken from the contract rather than measured — one target's frame is **8.68 GB**, so peak fell from about **26.6 GB to 9.8 GB per target**, roughly **16.8 GB** saved. For scale, run-0's OOM kill recorded `anon-rss:23778224kB` (#126); that incident's root cause was pandas on the *historical* leg and is not this, but the magnitude says this leg alone would have exhausted the same box. + +**The manager's historical frame is not the elephant, so it is not being chased.** #269 notes `_historical_frame` is held from `_read` through `_save`. By its own declared dimensions — 64,742 cells x 438 months = 28,356,996 rows — that is about **108 MB** at one float32 column, **1.2%** of a single forecast target frame. Recorded here rather than filed as its own issue, because a separate issue implying comparable cost would misdirect whoever picked it up. + +*Guarded.* `tests/test_track_a_source.py::test_shards_are_fetched_one_at_a_time_not_all_up_front` asserts the fetch/decode interleaving rather than a byte count — a memory threshold in a test is a flake on a busy machine, while "fetch, decode, fetch, decode" is exactly the property that bounds the peak. Mutation-proven: restoring the up-front dict produces `['fetch','fetch','fetch','decode','decode','decode']` and it fails. + +Cross-refs: **C-99** (the other defect the same delivery attempt found), **C-75** (the pandas retirement that #126 landed on the historical leg), views-postprocessing#269, views-postprocessing#126. + +--- + ### C-99: `_ContractStorePort.download` failed open where `upload` refuses — C-79's untreated sibling — RESOLVED | Field | Value | diff --git a/tests/test_track_a_source.py b/tests/test_track_a_source.py index 853f8b5..00427ab 100644 --- a/tests/test_track_a_source.py +++ b/tests/test_track_a_source.py @@ -68,7 +68,7 @@ def test_read_shard_round_trips_the_fixture(): def test_frames_for_target_assembles_the_run(): - frame, headers = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD}) + frame, headers = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD}.__getitem__) assert frame.n_rows == 6 and frame.sample_count == 4 # headers ride along in manifest shard order (provenance pass-through, §10.2) assert [h["time_id"] for h in headers] == [543] @@ -154,14 +154,14 @@ def test_minor_version_drift_accepted(): def test_missing_shard_bytes_rejected(): with pytest.raises(tas.TrackASourceError, match="not provided"): - tas.frames_for_target(MANIFEST, {}) + tas.frames_for_target(MANIFEST, {}.__getitem__) def test_shard_target_disagreeing_with_manifest_rejected(): bad = _retouched_shard(**{"metadata.json": _header(target="lr_ged_ns")}) manifest = {**MANIFEST, "shards": [{"name": SHARD_NAME, "sha256": _sha(bad)}]} with pytest.raises(tas.TrackASourceError, match="target"): - tas.frames_for_target(manifest, {SHARD_NAME: bad}) + tas.frames_for_target(manifest, {SHARD_NAME: bad}.__getitem__) def test_wrong_month_coverage_rejected(): @@ -170,21 +170,63 @@ def test_wrong_month_coverage_rejected(): # month 544's shard is absent entirely — caught at the bytes gate tas.frames_for_target( {**manifest, "shards": MANIFEST["shards"] + [{"name": "m544", "sha256": "0" * 64}]}, - {SHARD_NAME: SHARD}, + {SHARD_NAME: SHARD}.__getitem__, ) bad = _retouched_shard(**{"metadata.json": _header(time_id=999)}) manifest = {**MANIFEST, "shards": [{"name": SHARD_NAME, "sha256": _sha(bad)}]} with pytest.raises(tas.TrackASourceError, match="months covered"): - tas.frames_for_target(manifest, {SHARD_NAME: bad}) + tas.frames_for_target(manifest, {SHARD_NAME: bad}.__getitem__) def test_wrong_cell_count_rejected(): manifest = {**MANIFEST, "expected_cell_count": 7} with pytest.raises(tas.TrackASourceError, match="cells"): - tas.frames_for_target(manifest, {SHARD_NAME: SHARD}) + tas.frames_for_target(manifest, {SHARD_NAME: SHARD}.__getitem__) def test_manifest_missing_required_field_rejected(): truncated = {k: v for k, v in MANIFEST.items() if k != "expected_months"} with pytest.raises(tas.TrackASourceError, match="expected_months"): tas.read_manifest(json.dumps(truncated).encode()) + + +def test_shards_are_fetched_one_at_a_time_not_all_up_front(monkeypatch): + """The bound this function's memory shape depends on — register C-101. + + ``frames_for_target`` took a filled dict until 2026-08-14, so every shard of a + target was resident before the first was decoded. Measured at 36 shards, that plus + stacking with ``np.concatenate`` put peak at **3.06x the delivered frame**, in three + roughly equal thirds: the raw bytes, the per-shard frames, and the concatenated + copy. Fetching per shard and filling a manifest-sized buffer took it to **1.13x**. + + This asserts the *interleaving*, not a byte count, and deliberately so: a memory + threshold in a test is a flake on a busy machine, whereas "fetch, decode, fetch, + decode" is the property that actually bounds the peak and it is exactly observable. + Reverting to a pre-built dict makes the sequence fetch-fetch-decode-decode and this + fails; nothing else in the suite would notice. + """ + manifest = { + **MANIFEST, + "shards": [{"name": f"shard-{i}", "sha256": SHARD_SHA} for i in range(3)], + "expected_months": [543, 543, 543], + } + events = [] + real_read_shard = tas.read_shard + + def spy(shard_bytes, *, expected_sha256): + events.append("decode") + return real_read_shard(shard_bytes, expected_sha256=expected_sha256) + + monkeypatch.setattr(tas, "read_shard", spy) + + def fetch(name): + events.append("fetch") + return SHARD + + tas.frames_for_target(manifest, fetch) + + assert events == ["fetch", "decode"] * 3, ( + f"shards are not being fetched one at a time: {events}. Every 'fetch' that " + "precedes another 'fetch' is a shard's bytes held while the next is downloaded " + "— at 36 shards that was a third of the peak." + ) diff --git a/views_postprocessing/contract/track_a_source.py b/views_postprocessing/contract/track_a_source.py index 2aeeb80..5b18fb2 100644 --- a/views_postprocessing/contract/track_a_source.py +++ b/views_postprocessing/contract/track_a_source.py @@ -106,47 +106,83 @@ def read_shard(shard_bytes: bytes, *, expected_sha256: str) -> tuple[PredictionF def frames_for_target( - manifest: dict, shard_bytes_by_name: dict + manifest: dict, fetch_shard_bytes ) -> tuple[PredictionFrame, list[dict]]: """A (run, target)'s verified shards → ``(PredictionFrame, headers)``. - ``shard_bytes_by_name`` maps shard ``name`` → downloaded bytes; the manifest is the - only source of which shards exist (§3.3: names are locators, manifest content is - identity). Verifies run completeness against the manifest's own declarations — - months covered exactly, cell count per month — then stacks months into one frame. - The returned ``headers`` (manifest shard order) carry the producer-minted - provenance the sink passes through untouched (§10.2 — nothing is minted - downstream). + ``fetch_shard_bytes(name) -> bytes`` returns one shard's bytes on demand; the + manifest is the only source of which shards exist (§3.3: names are locators, + manifest content is identity). Verifies run completeness against the manifest's own + declarations — months covered exactly, cell count per month — then stacks months + into one frame. The returned ``headers`` (manifest shard order) carry the + producer-minted provenance the sink passes through untouched (§10.2 — nothing is + minted downstream). + + **A callback rather than a pre-built dict, and the output filled in place rather + than concatenated — both for memory (register C-101).** Measured on 2026-08-14 at + 36 shards: taking a dict meant every shard's bytes were resident before the first + was decoded, and stacking with ``np.concatenate`` meant the per-shard frames and + the finished array were resident together. Peak was **3.06x the delivered frame**, + in three roughly equal thirds. Fetching per shard and writing into a + manifest-sized buffer keeps one shard's bytes and one shard's frame alive at a + time, so peak is the frame plus a shard. + + The buffer is sized from ``expected_cell_count`` x shard count, which is a + declaration this function already enforces per shard — a shard whose row count + disagrees is refused *before* anything is written, so the slot arithmetic can + never straddle two months. """ - frames, months_seen, headers = [], [], [] - for entry in manifest["shards"]: + entries = manifest["shards"] + if not entries: + raise TrackASourceError( + "run: the manifest lists no shards — an empty run must not be assembled." + ) + expected_cells = manifest["expected_cell_count"] + values = time = unit = None + months_seen, headers = [], [] + + for position, entry in enumerate(entries): name = entry["name"] - if name not in shard_bytes_by_name: + try: + shard_bytes = fetch_shard_bytes(name) + except KeyError: raise TrackASourceError( f"run: manifest lists shard {name!r} but its bytes were not provided — " f"an unmanifested or missing shard must not be silently skipped." - ) - frame, header = read_shard(shard_bytes_by_name[name], expected_sha256=entry["sha256"]) + ) from None + frame, header = read_shard(shard_bytes, expected_sha256=entry["sha256"]) + # The bytes are dead the moment they are decoded. Dropping the reference here + # is the whole of the first third: without it every shard's bytes outlive the + # loop. + del shard_bytes if header.get("target") != manifest["target"]: raise TrackASourceError( f"run: shard header target {header.get('target')!r} != manifest target " f"{manifest['target']!r}." ) - if frame.n_rows != manifest["expected_cell_count"]: + if frame.n_rows != expected_cells: raise TrackASourceError( f"run: shard {name!r} carries {frame.n_rows} cells, manifest expects " - f"{manifest['expected_cell_count']}." + f"{expected_cells}." ) - frames.append(frame) + if values is None: + total = expected_cells * len(entries) + frame_time, frame_unit = np.asarray(frame.index.time), np.asarray(frame.index.unit) + values = np.empty((total, frame.values.shape[1]), dtype=frame.values.dtype) + time = np.empty(total, dtype=frame_time.dtype) + unit = np.empty(total, dtype=frame_unit.dtype) + start = position * expected_cells + stop = start + expected_cells + values[start:stop] = frame.values + time[start:stop] = np.asarray(frame.index.time) + unit[start:stop] = np.asarray(frame.index.unit) headers.append(header) months_seen.append(int(header.get("time_id"))) + del frame if sorted(months_seen) != sorted(int(m) for m in manifest["expected_months"]): raise TrackASourceError( f"run: months covered {sorted(months_seen)} != manifest expected " f"{sorted(manifest['expected_months'])} — a torn run must not be assembled." ) - values = np.concatenate([f.values for f in frames], axis=0) - time = np.concatenate([np.asarray(f.index.time) for f in frames]) - unit = np.concatenate([np.asarray(f.index.unit) for f in frames]) return build_prediction_frame(values, time, unit), headers diff --git a/views_postprocessing/contract/wire/source_selection.py b/views_postprocessing/contract/wire/source_selection.py index 92d4266..ac21cfc 100644 --- a/views_postprocessing/contract/wire/source_selection.py +++ b/views_postprocessing/contract/wire/source_selection.py @@ -64,12 +64,18 @@ def run_id(self) -> str: return self.manifest["run_id"] def load(self): - """Fetch (by pinned id), verify, curate — return the PRODUCT ``(frame, headers)``.""" - shard_bytes = { - name: self.store.download(file_id) - for name, file_id in self.shard_file_ids.items() - } - frame, headers = track_a_source.frames_for_target(self.manifest, shard_bytes) + """Fetch (by pinned id), verify, curate — return the PRODUCT ``(frame, headers)``. + + Shards are fetched **one at a time**, by handing ``frames_for_target`` a lookup + rather than a filled dict. The dict comprehension that stood here downloaded + every shard of the target before the first was decoded; with the stacking fix + beside it that made peak 3.06x the delivered frame (register C-101, measured). + Fetch-by-pinned-id is unchanged — the ids were pinned by ``resolve_run`` and a + newer run still cannot be mixed in. + """ + frame, headers = track_a_source.frames_for_target( + self.manifest, lambda name: self.store.download(self.shard_file_ids[name]) + ) for header in headers: found = header.get("provenance", {}).get("ensemble") if found != self.expected_ensemble: From e26fdeb88c639b09a0a484497c306edf2c62110e Mon Sep 17 00:00:00 2001 From: Polichinl Date: Fri, 14 Aug 2026 23:27:17 +0200 Subject: [PATCH 2/3] fix(wire): refuse mismatched draw counts in our own words, not numpy's MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review of the assembly change: sizing the buffer from the first shard MOVED a constraint. A run whose shards disagree on draws per cell used to fail inside np.concatenate; it now fails inside the assignment. Both are bare numpy ValueErrors — new: could not broadcast input array from shape (6,2) into shape (6,4) old: all the input array dimensions except for the concatenation axis must match exactly, but along dimension 1, the array at index 0 has size 4 ... neither names the shard, neither says "draws", and the second is only wordier. That is the C-99 shape exactly, caught before it could bite rather than after, and the constraint is now mine rather than incidental, so the refusal should be too. Mutation-proven: deleting the check brings the bare numpy error straight back. Suite 451 passed / 1 skipped / 39 xfailed, ruff clean. Co-Authored-By: Claude Opus 5 (1M context) --- reports/technical_risk_register.md | 2 ++ tests/test_track_a_source.py | 28 +++++++++++++++++++ .../contract/track_a_source.py | 11 ++++++++ 3 files changed, 41 insertions(+) diff --git a/reports/technical_risk_register.md b/reports/technical_risk_register.md index 5fb205b..45a2948 100644 --- a/reports/technical_risk_register.md +++ b/reports/technical_risk_register.md @@ -1152,6 +1152,8 @@ The ratio held at 12 and 36 shards, so it is the shape and not the scale. Fixing **The manager's historical frame is not the elephant, so it is not being chased.** #269 notes `_historical_frame` is held from `_read` through `_save`. By its own declared dimensions — 64,742 cells x 438 months = 28,356,996 rows — that is about **108 MB** at one float32 column, **1.2%** of a single forecast target frame. Recorded here rather than filed as its own issue, because a separate issue implying comparable cost would misdirect whoever picked it up. +*One refusal added, because the fix moved a constraint.* The buffer's width is fixed by the first shard, so a run whose shards disagree on draws per cell is now this function's constraint rather than an incidental one. Left to the assignment it surfaced as `could not broadcast input array from shape (6,2) into shape (6,4)` — no shard named, no mention of draws. The stacking it replaced was no better, only wordier. It now refuses in its own words, mutation-proven by deleting the check and watching the bare numpy error return. + *Guarded.* `tests/test_track_a_source.py::test_shards_are_fetched_one_at_a_time_not_all_up_front` asserts the fetch/decode interleaving rather than a byte count — a memory threshold in a test is a flake on a busy machine, while "fetch, decode, fetch, decode" is exactly the property that bounds the peak. Mutation-proven: restoring the up-front dict produces `['fetch','fetch','fetch','decode','decode','decode']` and it fails. Cross-refs: **C-99** (the other defect the same delivery attempt found), **C-75** (the pandas retirement that #126 landed on the historical leg), views-postprocessing#269, views-postprocessing#126. diff --git a/tests/test_track_a_source.py b/tests/test_track_a_source.py index 00427ab..0a3efd2 100644 --- a/tests/test_track_a_source.py +++ b/tests/test_track_a_source.py @@ -230,3 +230,31 @@ def fetch(name): "precedes another 'fetch' is a shard's bytes held while the next is downloaded " "— at 36 shards that was a third of the peak." ) + + +def test_shards_with_different_draw_counts_are_refused_in_our_own_words(): + """A run whose shards disagree on S is not one forecast — say so, do not let numpy. + + The assembly buffer's width is fixed by the first shard, which makes a draw-count + disagreement this function's constraint rather than an incidental one. Left to the + assignment it surfaces as ``could not broadcast input array from shape (6,2) into + shape (12,4)`` — no shard named, no mention of draws, three frames from anything a + reader recognises. The stacking it replaced was no better, only wordier; neither is + a refusal, which is the whole of C-99's lesson applied before it could bite again. + """ + narrow_values = io.BytesIO() + np.save(narrow_values, np.zeros((6, 2), dtype=np.float32)) + narrow = _retouched_shard(**{ + "y_pred.npy": narrow_values.getvalue(), + "metadata.json": _header(sample_count=2, time_id=544), + }) + manifest = { + **MANIFEST, + "shards": [ + {"name": SHARD_NAME, "sha256": SHARD_SHA}, + {"name": "narrow", "sha256": _sha(narrow)}, + ], + "expected_months": [543, 544], + } + with pytest.raises(tas.TrackASourceError, match="draws per cell"): + tas.frames_for_target(manifest, {SHARD_NAME: SHARD, "narrow": narrow}.__getitem__) diff --git a/views_postprocessing/contract/track_a_source.py b/views_postprocessing/contract/track_a_source.py index 5b18fb2..03f6609 100644 --- a/views_postprocessing/contract/track_a_source.py +++ b/views_postprocessing/contract/track_a_source.py @@ -171,6 +171,17 @@ def frames_for_target( values = np.empty((total, frame.values.shape[1]), dtype=frame.values.dtype) time = np.empty(total, dtype=frame_time.dtype) unit = np.empty(total, dtype=frame_unit.dtype) + elif frame.values.shape[1] != values.shape[1]: + # The buffer's width is fixed by the first shard, so a draw-count + # disagreement is now this function's constraint rather than numpy's. + # Left to the assignment it reads "could not broadcast input array from + # shape (a,b) into shape (a,c)" — no shard named, no mention of draws. + # Stacking said the same in more words; neither is a refusal (C-99). + raise TrackASourceError( + f"run: shard {name!r} carries {frame.values.shape[1]} draws per cell, " + f"the run's first shard carried {values.shape[1]} — a target assembled " + f"from shards with different sample counts is not one forecast." + ) start = position * expected_cells stop = start + expected_cells values[start:stop] = frame.values From 5868c685ea7ac89c35c56a6d976e9d5494f18881 Mon Sep 17 00:00:00 2001 From: Polichinl Date: Fri, 14 Aug 2026 23:44:54 +0200 Subject: [PATCH 3/3] fix(wire): the review found the rewrite's core was untested, and four false claims MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five reviewers on the assembly change. The memory result held — one reproduced it independently at 3.084x -> 1.158x against my 3.06x -> 1.13x, and separately confirmed that `del shard_bytes` really does free the bytes (np.load on a BytesIO copies; OWNDATA=True, base=None), that ordering is equivalent, and that array flags match np.concatenate exactly. What did not hold was the evidence I claimed for it. NOTHING IN THE SUITE ASSEMBLED MORE THAN ONE SHARD. I wrote that tests/test_wire_fixture.py passing unchanged proved the product had not changed. That file does not reference the assembly at all — zero mentions of frames_for_target, TargetLease or track_a_source. The real end-to-end proof is test_hop_b_sink_e2e.py, and its fixture has ONE shard, so `position * expected_cell_count` never ran above zero. The interleaving guard could not have covered it either: its three shards are byte-identical copies, so any ordering bug survives it. So the core of the rewrite shipped unverified. There is now a test that assembles three shards with distinct values, months and units and asserts equality with np.concatenate in manifest order. Mutation-proven three ways: reversing the slot index, an off-by-one in stop, and leaving the identifiers unwritten each fail it. A STORE FAULT WAS BEING BLAMED ON THE MANIFEST. frames_for_target reads a KeyError from the fetch callback as "this shard was never pinned", and my lambda put `self.store.download(...)` inside the same try. A KeyError from anywhere in the client — a response indexed with [] rather than .get(), which is exactly how C-99 happened one layer down — would have surfaced as "manifest lists shard X but its bytes were not provided", with `from None` discarding the traceback that said otherwise. Verified by mutation: the old lambda produces that false diagnosis verbatim. The lookup and the download are now separate, and a store KeyError raises a named SourceSelectionError chained to its cause. expected_cell_count SIZES AN ARRAY NOW. It used to sit on one side of a `!=`, where 6.0 compares equal to 6 and passed. It now reaches np.empty and raises a bare "TypeError: 'float' object cannot be interpreted as an integer", naming neither the field nor the manifest — from a document that crossed a repository boundary and whose read side only checks that the key is present. Guarded, with bool and non-positive covered. FOUR FALSE CLAIMS IN MY OWN PROSE. - "the residual is 2/n_shards": fits none of its own numbers. 2/12 is 0.17 against a measured 0.36. Re-measured at 12, 24 and 36 shards: the overhead is CONSTANT at about 4.5 shard-widths (~70 MB), so the ratio falls as 1/n — 1.36x, 1.19x, 1.13x. - "ADR-013 §5's ~1000 samples per cell": that line is under §0. §5 is the GAUL sidecar. - the test_wire_fixture citation, above. - a docstring quoting the numpy error as "into shape (12,4)" when the failing assignment touches a (6,4) slice. THE HISTORICAL FRAME IS FILED, NOT GLOSSED. #269 lists "the historical frame is released, or not held" as an acceptance criterion; this change does not meet it. It is now #273, carrying the measurement (108 MB, ~1.2% of a target frame) so nobody picks it up thinking it is comparable. Closing #269 with that quietly unmet was the alternative. Suite 457 passed / 1 skipped / 39 xfailed, ruff clean. Co-Authored-By: Claude Opus 5 (1M context) --- reports/technical_risk_register.md | 10 +- tests/test_track_a_source.py | 94 ++++++++++++++++++- tests/test_wire_source_selection.py | 38 ++++++++ .../contract/track_a_source.py | 11 +++ .../contract/wire/source_selection.py | 22 ++++- 5 files changed, 167 insertions(+), 8 deletions(-) diff --git a/reports/technical_risk_register.md b/reports/technical_risk_register.md index 45a2948..6e1f082 100644 --- a/reports/technical_risk_register.md +++ b/reports/technical_risk_register.md @@ -1142,15 +1142,17 @@ The 3x was **three roughly equal thirds**, and the issue named only the first: The ratio held at 12 and 36 shards, so it is the shape and not the scale. Fixing only (1), as the issue proposed, would have taken 3.06x to about 2x. -**The fix is one loop.** `frames_for_target` now takes `fetch_shard_bytes(name)` instead of a filled dict, drops each shard's bytes the moment they are decoded, and writes each shard into a manifest-sized buffer instead of stacking and concatenating. Peak is now the finished frame plus about one shard — the residual 0.13x is `2/n_shards`, which is why the 12-shard case measures 1.36x and the 36-shard case 1.13x. +**The fix is one loop.** `frames_for_target` now takes `fetch_shard_bytes(name)` instead of a filled dict, drops each shard's bytes the moment they are decoded, and writes each shard into a manifest-sized buffer instead of stacking and concatenating. Peak is now the finished frame plus a **fixed overhead of roughly 4.5 shard-widths** — the raw shard, its decoded array, and the intermediate copies `read_shard` makes unzipping and `np.load`-ing it. Because that overhead is constant while the frame grows with the shard count, the *ratio* falls as 1/n: measured 1.36x at 12 shards, 1.19x at 24, 1.13x at 36, with the absolute overhead steady at about 70 MB throughout. *(An earlier draft of this entry called the residual `2/n_shards`; review pointed out that fits none of its own numbers — 2/12 is 0.17 against a measured 0.36. Re-measured at three shard counts to get the constant above.)* *What makes the buffer safe.* Its slots are sized from `expected_cell_count`, a declaration this function already enforced per shard, and the enforcement runs **before** anything is written — so a shard whose row count disagrees is refused rather than straddling two months' slots. -*What proves the product did not change.* The wire is byte-frozen (ADR-013) and the golden fixture guards in `tests/test_wire_fixture.py` compare delivered artifacts against checked-in bytes. They pass unchanged, which is the claim: this is an assembly change with no output change. +*What proves the product did not change — corrected, because the first answer was wrong.* This entry originally cited `tests/test_wire_fixture.py`. **That file does not reference the assembly at all**: it round-trips static artifacts against checked-in bytes and never calls `frames_for_target` or `TargetLease.load`. The real end-to-end proof is `tests/test_hop_b_sink_e2e.py::test_e2e_byte_parity_with_the_fixture`, which drives the whole inbound chain and compares delivered bytes to the golden fixture — but **its fixture has one shard**, so `position * expected_cell_count` never ran with a position above zero. -**At production scale.** 64,742 cells x 36 months, at ADR-013 §5's *"~1000 samples per cell"* — the sample count is the one input here taken from the contract rather than measured — one target's frame is **8.68 GB**, so peak fell from about **26.6 GB to 9.8 GB per target**, roughly **16.8 GB** saved. For scale, run-0's OOM kill recorded `anon-rss:23778224kB` (#126); that incident's root cause was pandas on the *historical* leg and is not this, but the magnitude says this leg alone would have exhausted the same box. +The core of the rewrite was therefore unverified, and the interleaving guard could not have caught it either: its three shards are byte-identical copies, so any ordering bug would survive. `test_a_multi_shard_run_assembles_in_manifest_order_with_every_row_written` now assembles three shards with distinct values, months and units and asserts the result equals `np.concatenate` in manifest order. Mutation-proven three ways: reversing the slot index, an off-by-one in `stop`, and leaving the identifiers unwritten all fail it. -**The manager's historical frame is not the elephant, so it is not being chased.** #269 notes `_historical_frame` is held from `_read` through `_save`. By its own declared dimensions — 64,742 cells x 438 months = 28,356,996 rows — that is about **108 MB** at one float32 column, **1.2%** of a single forecast target frame. Recorded here rather than filed as its own issue, because a separate issue implying comparable cost would misdirect whoever picked it up. +**At production scale.** 64,742 cells x 36 months, at ADR-013 **§0**'s *"~1000 samples per cell"* — the sample count is the one input here taken from the contract rather than measured — one target's frame is **8.68 GB**, so peak fell from about **26.6 GB to 9.8 GB per target**, roughly **16.8 GB** saved. For scale, run-0's OOM kill recorded `anon-rss:23778224kB` (#126); that incident's root cause was pandas on the *historical* leg and is not this, but the magnitude says this leg alone would have exhausted the same box. + +**The manager's historical frame is not the elephant, so it is not being chased.** #269 notes `_historical_frame` is held from `_read` through `_save`. By its own declared dimensions — 64,742 cells x 438 months = 28,356,996 rows — that is about **108 MB** at one float32 column, **1.2%** of a single forecast target frame. **Filed as #273**, carrying the measurement so it cannot be picked up under the impression that it is comparable — and because #269 listed *"the historical frame is released, or not held"* as an acceptance criterion of its own, which this change does not meet. Closing #269 while quietly leaving that unmet was the alternative, and it is not one. *One refusal added, because the fix moved a constraint.* The buffer's width is fixed by the first shard, so a run whose shards disagree on draws per cell is now this function's constraint rather than an incidental one. Left to the assignment it surfaced as `could not broadcast input array from shape (6,2) into shape (6,4)` — no shard named, no mention of draws. The stacking it replaced was no better, only wordier. It now refuses in its own words, mutation-proven by deleting the check and watching the bare numpy error return. diff --git a/tests/test_track_a_source.py b/tests/test_track_a_source.py index 0a3efd2..d96ae68 100644 --- a/tests/test_track_a_source.py +++ b/tests/test_track_a_source.py @@ -238,7 +238,7 @@ def test_shards_with_different_draw_counts_are_refused_in_our_own_words(): The assembly buffer's width is fixed by the first shard, which makes a draw-count disagreement this function's constraint rather than an incidental one. Left to the assignment it surfaces as ``could not broadcast input array from shape (6,2) into - shape (12,4)`` — no shard named, no mention of draws, three frames from anything a + shape (6,4)`` — no shard named, no mention of draws, three frames from anything a reader recognises. The stacking it replaced was no better, only wordier; neither is a refusal, which is the whole of C-99's lesson applied before it could bite again. """ @@ -258,3 +258,95 @@ def test_shards_with_different_draw_counts_are_refused_in_our_own_words(): } with pytest.raises(tas.TrackASourceError, match="draws per cell"): tas.frames_for_target(manifest, {SHARD_NAME: SHARD, "narrow": narrow}.__getitem__) + + +def _shard_with(*, values, time_id, unit_start): + """A fixture-shaped shard carrying declared values/ids — distinct per month.""" + payload, ids = io.BytesIO(), io.BytesIO() + np.save(payload, values) + t, u = io.BytesIO(), io.BytesIO() + np.save(t, np.full(values.shape[0], time_id, dtype=np.int64)) + np.save(u, np.arange(unit_start, unit_start + values.shape[0], dtype=np.int64)) + with zipfile.ZipFile(ids, "w", zipfile.ZIP_STORED) as zf: + zf.writestr("time.npy", t.getvalue()) + zf.writestr("unit.npy", u.getvalue()) + return _retouched_shard(**{ + "y_pred.npy": payload.getvalue(), + "identifiers.npz": ids.getvalue(), + "metadata.json": _header(sample_count=values.shape[1], time_id=time_id), + }) + + +def test_a_multi_shard_run_assembles_in_manifest_order_with_every_row_written(): + """The slot arithmetic, against the stacking it replaced — register C-101. + + Until this existed, **nothing in the suite assembled more than one shard**. The + single-shard fixture drives the whole e2e byte-parity chain + (`tests/test_hop_b_sink_e2e.py`), so `position * expected_cell_count` never ran with + a position above zero, and the interleaving guard's three shards are byte-identical + copies that would survive any ordering bug. The rewrite's core was unverified. + + Three shards with values, months and units that are distinct per shard, asserted + against exactly what `np.concatenate` in manifest order produces. That is the claim + the change makes: same product, less memory. It catches a transposed slot, an + off-by-one in `start`/`stop`, an unwritten row left as `np.empty` garbage, and + identifiers assembled out of step with the values they label. + """ + cells, draws = 4, 3 + blocks = [ + np.full((cells, draws), fill, dtype=np.float32) for fill in (1.5, 2.5, 3.5) + ] + shards, entries, months = {}, [], [] + for i, block in enumerate(blocks): + time_id = 543 + i + raw = _shard_with(values=block, time_id=time_id, unit_start=100_000 + 10 * i) + name = f"m{time_id}" + shards[name] = raw + entries.append({"name": name, "sha256": _sha(raw)}) + months.append(time_id) + + manifest = { + **MANIFEST, + "shards": entries, + "expected_months": months, + "expected_cell_count": cells, + } + frame, headers = tas.frames_for_target(manifest, shards.__getitem__) + + np.testing.assert_array_equal(frame.values, np.concatenate(blocks, axis=0)) + np.testing.assert_array_equal( + np.asarray(frame.index.time), + np.concatenate([np.full(cells, m, dtype=np.int64) for m in months]), + ) + np.testing.assert_array_equal( + np.asarray(frame.index.unit), + np.concatenate([ + np.arange(100_000 + 10 * i, 100_000 + 10 * i + cells, dtype=np.int64) + for i in range(len(blocks)) + ]), + ) + assert frame.n_rows == cells * len(blocks) + assert [h["time_id"] for h in headers] == months, "headers ride in manifest order" + + +@pytest.mark.parametrize( + "declared, why", + [ + (6.0, "a JSON float compares equal to 6 and used to pass"), + (0, "zero cells sizes an empty frame nothing can be checked against"), + (-1, "negative would raise deep inside numpy"), + (True, "bool is an int subclass and would size a one-row frame"), + ], +) +def test_expected_cell_count_must_be_a_positive_integer(declared, why): + """It sizes an array now; it used to sit on one side of a ``!=``. + + ``6.0 == 6`` is True, so a manifest carrying a JSON float passed the old row-count + check and assembled correctly. The rewrite hands the same value to ``np.empty``, + where it raises ``TypeError: 'float' object cannot be interpreted as an integer`` — + bare, naming neither the field nor the manifest, from a document that crossed a + repository boundary. ``read_manifest`` only checks that the key is present. + """ + manifest = {**MANIFEST, "expected_cell_count": declared} + with pytest.raises(tas.TrackASourceError, match="expected_cell_count"): + tas.frames_for_target(manifest, {SHARD_NAME: SHARD}.__getitem__) diff --git a/tests/test_wire_source_selection.py b/tests/test_wire_source_selection.py index accac29..0d84e23 100644 --- a/tests/test_wire_source_selection.py +++ b/tests/test_wire_source_selection.py @@ -141,3 +141,41 @@ def test_selection_filters_are_golden_strings(): assert sel.HOP_A_SHARD_FILTERS == {"category": "forecast", "type": "sampled_forecast_shard"} assert sel.HOP_A_MANIFEST_FILTERS == {"category": "forecast", "type": "sampled_forecast_manifest"} assert sel.HOP_A_MANIFEST_NAME_TEMPLATE == "{run_id}__{target}__manifest.json" + + +def test_a_store_that_raises_keyerror_is_not_blamed_on_the_manifest(): + """A store fault must not be relabelled as a missing manifest entry. + + ``frames_for_target`` reads a ``KeyError`` from the fetch callback as *"this shard + was never pinned"*. The lease's callback calls into the store, so before this guard + a ``KeyError`` thrown anywhere inside the client — a response shape indexed with + ``[]`` rather than ``.get()``, which is how C-99 happened one layer down — would + surface as *"manifest lists shard X but its bytes were not provided"*, blaming the + manifest for a store failure. ``raise ... from None`` would have discarded the + traceback that said otherwise. + + The shard here IS pinned, so "not provided" would be a false diagnosis. + """ + manifest = json.loads((_FIX / _MANIFEST_NAME).read_text()) + + class KeyErroringStore: + def download(self, file_id): + raise KeyError("data") + + lease = sel.TargetLease( + target=manifest["target"], + manifest=manifest, + shard_file_ids={entry["name"]: "pinned-id" for entry in manifest["shards"]}, + store=KeyErroringStore(), + expected_ensemble="fixture_ensemble", + ) + with pytest.raises(sel.SourceSelectionError) as excinfo: + lease.load() + message = str(excinfo.value) + assert "store fault" in message, "the refusal must say where the fault is" + assert "not provided" not in message, ( + "a store KeyError must not be reported as a missing manifest entry" + ) + assert excinfo.value.__cause__ is not None, ( + "the store's own KeyError must be chained, not discarded" + ) diff --git a/views_postprocessing/contract/track_a_source.py b/views_postprocessing/contract/track_a_source.py index 03f6609..416f0a8 100644 --- a/views_postprocessing/contract/track_a_source.py +++ b/views_postprocessing/contract/track_a_source.py @@ -138,6 +138,17 @@ def frames_for_target( "run: the manifest lists no shards — an empty run must not be assembled." ) expected_cells = manifest["expected_cell_count"] + # It sizes an array now, where it used to sit on one side of a `!=`. `6.0` compares + # equal to `6` and passed the old check happily; here it reaches `np.empty` and + # raises a bare `TypeError: 'float' object cannot be interpreted as an integer`, + # naming neither the manifest nor the field. `read_manifest` checks that the key is + # present, never what it holds, and the manifest crosses a repository boundary. + if type(expected_cells) is not int or expected_cells < 1: + raise TrackASourceError( + f"run: manifest declares expected_cell_count={expected_cells!r} " + f"({type(expected_cells).__name__}) — it must be a positive integer, " + f"because it sizes the assembled frame." + ) values = time = unit = None months_seen, headers = [], [] diff --git a/views_postprocessing/contract/wire/source_selection.py b/views_postprocessing/contract/wire/source_selection.py index ac21cfc..25a79bc 100644 --- a/views_postprocessing/contract/wire/source_selection.py +++ b/views_postprocessing/contract/wire/source_selection.py @@ -73,9 +73,25 @@ def load(self): Fetch-by-pinned-id is unchanged — the ids were pinned by ``resolve_run`` and a newer run still cannot be mixed in. """ - frame, headers = track_a_source.frames_for_target( - self.manifest, lambda name: self.store.download(self.shard_file_ids[name]) - ) + def fetch(name): + # `frames_for_target` reads a KeyError as "this shard was never pinned". + # Only the lookup is allowed to say that: a KeyError thrown from inside the + # store — a response shape indexed with [] somewhere in the client — would + # otherwise be relabelled "bytes were not provided", blaming the manifest + # for a store failure and discarding the traceback that says otherwise. + # C-99 was that exact substitution one layer down. + file_id = self.shard_file_ids[name] + try: + return self.store.download(file_id) + except KeyError as exc: + raise SourceSelectionError( + f"run {self.run_id!r}, target {self.target!r}: the store raised " + f"KeyError({exc}) downloading shard {name!r} (file_id {file_id!r}). " + f"The shard was pinned and requested — this is a store fault, not a " + f"missing manifest entry." + ) from exc + + frame, headers = track_a_source.frames_for_target(self.manifest, fetch) for header in headers: found = header.get("provenance", {}).get("ensemble") if found != self.expected_ensemble: