#269: assembling a target held three copies of it, not one — 3.06x peak to 1.13x

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,51 @@ 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 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 — 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.
+
+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.
+
+**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.
+
+*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 |

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,183 @@ 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."
+    )
+
+
+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 (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.
+    """
+    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__)
+
+
+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__)

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"
+    )