diff --git a/deployment/aws/lambda_handler.py b/deployment/aws/lambda_handler.py index b38867d..7857f0e 100644 --- a/deployment/aws/lambda_handler.py +++ b/deployment/aws/lambda_handler.py @@ -441,6 +441,7 @@ def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: Default ``mode`` (or no mode) runs per-cell processing. ``mode="setup"`` creates the zarr template; ``mode="finalize"`` consolidates metadata; + ``mode="coverage"`` writes the store-root ``coverage.moc`` (issue #200); ``mode="extract"`` extracts chunk-boundary geometry parquets (issue #148); ``mode="process_event"`` runs the temporal/event worker (issue #12). """ @@ -455,6 +456,8 @@ def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: return _handle_setup(event) if mode == "finalize": return _handle_finalize(event) + if mode == "coverage": + return _handle_coverage(event) # Extract mode returns directly: the result_url mirror below is for the # per-unit fan-out handlers (spatial process, temporal process_event) only. if mode == "extract": @@ -673,6 +676,36 @@ def _handle_finalize(event: Dict[str, Any]) -> Dict[str, Any]: return {"statusCode": 500, "body": json.dumps({"error": str(e), "mode": "finalize"})} +def _handle_coverage(event: Dict[str, Any]) -> Dict[str, Any]: + """Write/union the store-root ``coverage.moc`` (issue #200 phase 3). + + Posted fire-and-forget (``InvocationType="Event"``) by the dispatcher at + end of run: the orchestrator can compute the shard-order MOC but cannot + PUT to S3, so the SERIALIZED envelope rides in the event (bounded by + construction — see the dispatch-site comment in ``zagg.runner``) and the + worker GET-unions-PUTs one root object. Nobody reads this response on + the Event invoke; errors are logged and fail open — the root MOC is a + regenerable cache (D9): readers degrade to the sweep MOC or the walk, + never to wrong answers. + """ + from zagg.hive import write_root_coverage + + logger.info(f"Coverage mode: writing root coverage.moc at {event.get('store_path')}") + try: + merged = write_root_coverage( + event["store_path"], event["coverage"], **_output_store_kwargs(event) + ) + return { + "statusCode": 200, + "body": json.dumps( + {"ok": True, "mode": "coverage", "ranges": len(merged.get("ranges", []))} + ), + } + except Exception as e: + logger.exception(e) + return {"statusCode": 500, "body": json.dumps({"error": str(e), "mode": "coverage"})} + + def _json_scalar(v: Any) -> Any: """Coerce one result value to a JSON-safe scalar (numpy float -> float).""" if v is None: diff --git a/docs/design/sparse_coverage.md b/docs/design/sparse_coverage.md index a9b0fe3..e7ec096 100644 --- a/docs/design/sparse_coverage.md +++ b/docs/design/sparse_coverage.md @@ -127,19 +127,21 @@ the data itself). Tracks [#200](https://github.com/englacial/zagg/issues/200). is frozen with the mortie-side spec (O8). Future *store-level* covers that cross base cells generalize to **≤ 12 members** (the 12 base cells). Readers AOI-reject on the box without parsing anything larger. -- **Tier 1 — the budgeted shard MOC**: cell-level coverage, exact if it fits - the budget, else **coarsened until it fits**. Coarsening only over-covers - (conservative superset): false positives cost one wasted leaf read; false - negatives are impossible. The achieved depth is recorded with the payload. - Budget scale is open (O8) and is matched to **S3/cloud conventions, not - database-page conventions** — candidates span KB to low MB (the PostgreSQL - ~2 KB TOAST inline threshold is retained only as a portability footnote). - The budget also picks the carrier: a KB-scale payload rides the commit - stamp attrs (zero extra objects); a hundreds-of-KB tier becomes its own - object inside the leaf, with only the box + achieved depth + pointer in - attrs (one extra GET, paid only by readers that want it). If the in-leaf - object survives O8, it is a *noted exception* to §2's "vanilla zarr v3" - leaf: one foreign key inside the leaf, invisible to zarr readers. +- **Tier 1 — the exact shard bitmap** *(as O8 resolved it — the originally + drafted "budgeted, coarsen-to-fit MOC" was superseded by the + [#202 item (6) measurement](https://github.com/englacial/zagg/issues/200#issuecomment-4939264286): + for linear-track occupancy, coarsen-to-fit ranges at KB budgets deliver + only box-level filtering while a compressed bitmap reaches exact at + ~25 KB)*: the shard's cell-order occupancy as a **zstd-compressed bit + field**, one bit per subtree cell in ascending packed-word order, stored + as the in-leaf `coverage.moc` sidecar. Raw size is deterministic + (`ceil(4^depth/8)`) regardless of fragmentation; no coarsened variant is + built (one code path, sidecar-only). The stamp attrs carry only the box + + the bitmap's order + pointer + byte sizes (one extra GET, paid only by + readers that pass the box test). The sidecar is the *noted exception* to + §2's "vanilla zarr v3" leaf: one foreign key inside the leaf, ignored by + zarr readers (data reads unaffected; member enumeration warns and skips + it). - **Tier 2 — exact**: the `morton` coordinate array in the leaf *is* the exact cell list. The MOC tiers are indexes, never truth (D9 discipline, applied one level down). @@ -171,10 +173,13 @@ the data itself). Tracks [#200](https://github.com/englacial/zagg/issues/200). cost effectively nothing and answer the actual question readers ask ("where is there data?") in one GET. -Open items: budget scale / serialization / carrier / pad sentinel for the -shard MOC tiers (O8, measured on -[#202](https://github.com/englacial/zagg/issues/202) item (6)); the -end-of-run root MOC default (O9); root-object serialization format (O1). +O8 and O9 are **resolved** (espg-ratified on the +[#200 thread](https://github.com/englacial/zagg/issues/200#issuecomment-4939477871), +implemented on PR #208): the shard tier is an **exact cell-order occupancy +bitmap, zstd-compressed, as an in-leaf sidecar object** — attrs carry only +the tier-0 box + order + pointer + sizes, with a null pad sentinel — and the +end-of-run root MOC **defaults on** for hive stores. The root object +serializes per O1 as JSON ranges with decimal-string endpoints. ## 5. Layer 3: reader architecture @@ -323,17 +328,31 @@ write path (§2) is load-bearing; this phase is optimization. doesn't list), regenerate explicitly (`refresh=True` / the sweep); no wall-clock staleness horizon. The incremental-run half is settled by the §4 lifecycle (leaves are durable truth; the sweep rebuilds correctly). -- **O8 — shard-MOC budget, serialization, carrier, pad sentinel**: byte - scale matched to S3/cloud conventions (candidates KB → low MB); ranges - vs. bitmap serialization; commit-stamp attrs payload vs. in-leaf object - carrier; box pad sentinel (base-0/null vs. repetition). Decided using - the #202 item (6) measurement (SERC + 88S shards; the tier-0 box is the - baseline each budget must beat; over-coverage reported in STAC - simplification-error terms). Frozen alongside the mortie spec. -- **O9 — end-of-run root MOC default**: on (the fire-and-forget Event - invoke is fail-open and run-size independent) vs. off until #202's - full-AOI runs record the invoke round trip. Either is a one-line flag; - espg leaning cautious pending the benchmark-impact numbers. + Implemented in this shape by PR #208's reader primitives + (`zagg.coverage`: `warn_if_stale` — once per store, never auto-walk — + and `refresh_root_coverage`, the explicit walk); the concurrent-run + GET-union-PUT race (last writer wins until re-union/sweep) is recorded + there as accepted under this same lean. +- **O8 — shard-MOC budget, serialization, carrier, pad sentinel**: + **RESOLVED** ([espg-ratified](https://github.com/englacial/zagg/issues/200#issuecomment-4939477871), + from the [#202 item (6) measurement](https://github.com/englacial/zagg/issues/200#issuecomment-4939264286)): + no budgeted/coarsened tier — the leaf encoding is an **exact cell-order + bitmap, zstd-compressed, as the in-leaf `coverage.moc` sidecar** (raw size + deterministic at `ceil(4^depth/8)`, immune to the ragged worst case where + exact ranges hit MB-scale on linear-track occupancy at ~1.1 cells/range). + Stamp attrs carry only the tier-0 box + the bitmap's order + pointer + + byte sizes; pad sentinel is JSON null; the envelope gains an + `encoding: "ranges" | "bitmap"` discriminator. Bit convention (frozen + with the mortie spec, golden-vector-pinned on PR #208): bit i = the i-th + shard-subtree cell in ascending packed-word order (base-4 D1 digit tail, + digits 1..4 → 0..3), MSB-first per byte. +- **O9 — end-of-run root MOC default**: **RESOLVED — on** for hive stores + ([espg](https://github.com/englacial/zagg/issues/200#issuecomment-4938859764): + coverage MOCs are the default for healpix templates; PR #208 implements + `output.coverage_moc`, default true under `store_layout: hive`, explicit + true rejected elsewhere). The write is fail-open on both backends; the + Lambda leg is one fire-and-forget `mode: "coverage"` Event invoke with the + pre-serialized ranges envelope. ## 9. References diff --git a/docs/hive_layout.md b/docs/hive_layout.md index 4504625..ebf9463 100644 --- a/docs/hive_layout.md +++ b/docs/hive_layout.md @@ -27,11 +27,16 @@ round — wired to the **local backend only** (see [Status](#status)). naturally: every order is a legal node. - **Full id at the leaf** (D3): `.../-5/1/1/2/3/3/3/-5112333.zarr` is self-describing without parsing its directory chain — greppable in - inventories, unambiguous if moved. Each leaf is a completely vanilla zarr v3 + inventories, unambiguous if moved. Each leaf is a vanilla zarr v3 store: the same group/array template as the flat layout, sized to one shard (dense arrays hold `cells_per_shard` cells; `resolution: chunk` companions hold the shard's K inner chunks; CSR ragged subgroups sit under the same - group path, named by the shard label at K==1). + group path, named by the shard label at K==1). One recorded exception + ([issue #200](https://github.com/englacial/zagg/issues/200), O8): the + `coverage.moc` occupancy-bitmap sidecar inside the leaf — a single foreign + key that zarr readers ignore (data reads are unaffected; member + enumeration like `members()`/`tree()` emits a `ZarrUserWarning` and skips + it). - **Node invariant** (D5): below the root, a node contains *only* digit children (`[1-4]/`) and `*.zarr` objects — zero zarr metadata above the leaf, no shared mutable state across workers. The root alone also carries @@ -120,21 +125,137 @@ template creates anyway. A shard that errors, or streams no chunks (no data), leaves no stamp; a fully empty shard leaves no `.zarr/` prefix at all (the leaf is created lazily on the first chunk write). +The stamp also carries the shard's **coverage envelope** — see +[Coverage](#coverage) below. The sidecar it points to is written before the +stamp, so coverage shares the debris semantics: no stamp, no visible coverage. + +## Coverage + +Where the data is, declared hierarchically +([issue #200](https://github.com/englacial/zagg/issues/200), design §4 as +amended by PR #206; O8/O9 resolved on the issue thread). Three tiers per +shard plus one store-root object: + +| tier | what | where | cost to read | +|---|---|---|---| +| 0 — morton box | canonical ≤ 4-member cover of the occupied cells (DCA children, each tightened) | `coverage` payload on the commit stamp | free — rides the stamp GET readers already make | +| 1 — exact bitmap | zstd-compressed bit field over the shard subtree at `cell_order` | `{full_id}.zarr/coverage.moc` sidecar | one opt-in GET | +| 2 — exact truth | the leaf's `morton` coordinate array | the leaf's data plane | array read; the tiers above are indexes, never truth (D9) | +| root | shard-order ranges MOC over all completed shards | `{store_root}/coverage.moc` | one GET — the discovery bootstrap | + +**Leaf envelope** (on the stamp, `zagg.hive.read_coverage`; strict +`spec: morton-moc/1` gate — unknown specs read as absent): + +```json +"coverage": { + "spec": "morton-moc/1", + "box": ["-42113221", "-42113224", null, null], + "cell_order": 12, + "source": "worker", + "encoding": "bitmap", + "sidecar": "coverage.moc", + "nbytes": 213, + "raw_nbytes": 512 +} +``` + +`box` is always exactly 4 slots, nulls trailing; members are D1 decimal +strings. `encoding`/`sidecar`/sizes appear only when the bitmap exists — a +box-only envelope (phase-1-era leaf, or a depth-0 `child_order == +parent_order` config) is read as "box only". No `generated_at`: the stamp's +`written_at` is the one clock and one writer. Bit convention (frozen with +the mortie-side spec): bit i = the i-th shard-subtree cell in ascending +packed-word order (base-4 value of the D1 digit tail, digits 1..4 → 0..3), +MSB-first per byte. A corrupt sidecar (bad zstd, wrong size) **raises**; a +missing one degrades to `None` — a truncated bitmap must never read as a +plausible partial cell set. The sidecar is the one foreign key inside the +otherwise-vanilla leaf: zarr data reads are unaffected, but member +enumeration (`members()`/`tree()`) emits a `ZarrUserWarning` and skips it. + +**Root envelope** (`{store_root}/coverage.moc`, `zagg.coverage.load_coverage`): + +```json +{ + "spec": "morton-moc/1", + "encoding": "ranges", + "order": 6, + "source": "dispatcher", + "generated_at": "2026-07-10T22:59:35+00:00", + "ranges": [["5112333", "5112333"], ["-4211321", "-4211324"]] +} +``` + +The example above is `zagg.hive.build_root_coverage` output for the shards +`-4211321..-4211324` plus `5112333` (all order 6) and round-trips through +`root_coverage_words`; the test suite parses it straight out of this file so +the reference example can never drift from the implementation. + +A range is an inclusive run of same-order cells within one base cell, +consecutive in digit-tail rank; endpoints are decimal **strings** (packed +u64 words exceed 2^53 and raw JSON numbers get mangled by float-based +parsers). `source` is `"dispatcher"` (end-of-run write) or `"refresh"` (the +explicit walk rebuild); the sweep will add its own. + +**Reader flow** (`zagg.coverage`): `load_coverage` → `root_coverage_and` +against the AOI to pick candidate shards (one GET, no walk); per leaf, +`box_and` on the stamp payload for the cheap reject, then `bitmap_and` for +exact cell-level filtering (falls back to the box verdict with `None` when +the leaf is box-only), then the `morton` coordinate as truth. The box is a +conservative superset — false positives cost one wasted read, false +negatives are impossible; the bitmap and the root MOC are exact for what +they list. + +**Staleness (O7)**: readers trust silently on the hot path. The root object +is written fail-open at **end of run** while leaves stamp continuously, so +the most common gap is benign — a run still in progress. Beyond that, a +crashed run, an out-of-band write, or the benign concurrent-run union race +(GET-union-PUT is not atomic; last writer wins until the next re-union) +leaves it missing shards, which degrades to "reader doesn't see the newest +run", never a wrong answer. `zagg.coverage.warn_if_stale` implements the +lazy detection lean: when a reader opens a commit-stamped leaf the root MOC +doesn't list, it warns once per store and suggests +`zagg.coverage.refresh_root_coverage` — the explicit delimiter-LIST walk +that rebuilds the root MOC from the stamped leaves (debris excluded) and +writes it with `source: "refresh"`. No reader ever auto-walks (D10). + +**Deploy note** (mirrors the PR #205 setup-echo note): the Lambda leg posts +one fire-and-forget `mode: "coverage"` invoke, which requires the +redeployed function. An **older deployment 400s the event in its process +handler** — a logged error line in CloudWatch, but no writes, no result +mirror, and no async redelivery — so the failure is fail-open by +construction; the root object simply doesn't appear until the sweep or a +refresh builds it. + ## Reading a hive store There is no store-root `zarr.open()` (deliberately — D12; a root hierarchy can be added later by the sweep as a derived artifact). Readers: 1. GET `morton_hive.json` (once, cacheable) → `shard_order`, `cell_order`. -2. Compute a shard's leaf path by string arithmetic on its decimal id +2. GET `coverage.moc` (`zagg.coverage.load_coverage`) → the covered shard + set, intersected with the AOI (`root_coverage_and`) — see + [Coverage](#coverage). +3. Compute a shard's leaf path by string arithmetic on its decimal id (`zagg.hive.shard_leaf_path`), open the leaf zarr, and **check the commit - stamp** (`zagg.hive.read_commit`) before trusting the contents. -3. Discovery without a shard list is a delimiter-LIST walk: recurse on - `[1-4]/` children; a `*.zarr` entry is data at that node; no digit children - ⇒ nothing finer. Never LIST per observation in a join loop (D10). + stamp** (`zagg.hive.read_commit`) before trusting the contents; the + stamp's coverage payload pre-filters the AOI (`box_and`/`bitmap_and`). +4. Discovery without a root MOC falls back to the delimiter-LIST walk: + recurse on `[1-4]/` children; a `*.zarr` entry is data at that node; no + digit children ⇒ nothing finer. Never LIST per observation in a join + loop (D10). -The `coverage.moc` domain declaration (§4 of the design record) is a follow-on -issue and will remove the walk from the discovery path too. +The store-root `coverage.moc` ([issue #200](https://github.com/englacial/zagg/issues/200) +phase 3, default-on for hive) removes the walk from the bootstrap path: one +GET of the root object yields the shard-order coverage MOC (JSON ranges, +decimal-string endpoints). It is written fail-open at end of run — by the +dispatcher directly (local) or one fire-and-forget `mode: "coverage"` worker +invoke (Lambda; an older deployment has no coverage mode and 400s the event +in its process handler — logged, no writes, no async retry — which is safe: +the object is a regenerable cache under D9, and readers degrade to the walk). +Incremental runs union with the existing object; concurrent runs race +benignly (GET-union-PUT is not atomic: last writer wins, and its union may +miss the loser's shards until the sweep or the next run re-unions — accepted +under D9/O7). The §7 sweep remains the authoritative rebuilder. ## Status @@ -146,7 +267,12 @@ issue and will remove the walk from the discovery path too. the event config's orders, emits its own leaf template, and stamps completion as its final PUT. The async status channel stays at the flat sibling prefix (`{store_root}.status//…`), outside the digit tree. -- The store-level `coverage.moc` (and any per-shard MOC stamping) waits on - the design in [issue #200](https://github.com/englacial/zagg/issues/200). +- **Coverage ships** ([issue #200](https://github.com/englacial/zagg/issues/200) + phases 1–4): the tier-0 morton box on the commit stamp, the exact + zstd-bitmap `coverage.moc` sidecar inside each leaf, the end-of-run + store-root `coverage.moc` (shard-order ranges MOC, `output.coverage_moc`, + default on for hive) for the one-GET bootstrap, plus the `zagg.coverage` + reader primitives (per-tier AOI intersection, O7 staleness lean, explicit + refresh). - Write-throughput validation at fleet scale is tracked with the benchmark machinery in [issue #202](https://github.com/englacial/zagg/issues/202). diff --git a/src/zagg/config.py b/src/zagg/config.py index f75127d..66aad2f 100644 --- a/src/zagg/config.py +++ b/src/zagg/config.py @@ -345,6 +345,28 @@ def validate_config(config: PipelineConfig) -> None: "consolidate (D5/D12) — drop output.consolidate_metadata" ) + # End-of-run root coverage MOC (issue #200 phase 3), boolean when present. + # Default ON for the hive layout (O9: coverage is the default on healpix + # templates); it writes {store}/coverage.moc, a hive-root object, so an + # EXPLICIT true on a non-healpix grid or a flat-layout store is a config + # mistake, not a no-op — reject it pointedly. Absent simply means off + # there (get_coverage_moc resolves the default). + coverage_moc = config.output.get("coverage_moc") + if coverage_moc is not None and not isinstance(coverage_moc, bool): + raise ValueError(f"output.coverage_moc must be a boolean (got {coverage_moc!r})") + if coverage_moc: + if (grid or {}).get("type", "healpix") != "healpix": + raise ValueError( + "output.coverage_moc requires a healpix grid (the root coverage.moc " + "is a morton MOC; drop the flag for rectilinear output)" + ) + if store_layout != "hive": + raise ValueError( + "output.coverage_moc requires output.store_layout: hive (the root " + "coverage.moc lives at the hive store root; flat stores have no " + "hive root to bootstrap from)" + ) + # Validate bounds structure (optional) if config.bounds is not None: allowed_keys = {"temporal", "spatial"} @@ -1554,6 +1576,22 @@ def get_store_layout(config: PipelineConfig) -> str: return config.output.get("store_layout") or "flat" +def get_coverage_moc(config: PipelineConfig) -> bool: + """Whether the end-of-run root ``coverage.moc`` write is on (issue #200). + + Default ON for hive-layout stores (O9/espg: coverage MOCs are the default + for healpix templates — and hive is healpix-only by validation); + ``output.coverage_moc: false`` opts out. Flat-layout / non-healpix + configs default off, and an explicit ``true`` there is rejected by + ``validate_config``. A present-but-null key falls back to the default, + like ``store_layout``. + """ + flag = config.output.get("coverage_moc") + if flag is None: + return get_store_layout(config) == "hive" + return bool(flag) + + def get_aoi_mask(config: PipelineConfig) -> bool: """Whether the optional strict-AOI cell mask is enabled (issue #101). diff --git a/src/zagg/coverage.py b/src/zagg/coverage.py new file mode 100644 index 0000000..c73ffee --- /dev/null +++ b/src/zagg/coverage.py @@ -0,0 +1,271 @@ +"""Reader-side coverage primitives — issue #200 phase 4 (design §5-lite). + +The minimal consumption layer over the write-side machinery in +:mod:`zagg.hive` (the full §5 reader architecture is its own effort): load +the store-root envelope, intersect an AOI against each coverage tier (root +ranges MOC → leaf tier-0 box → leaf bitmap sidecar → the leaf's ``morton`` +coordinate as exact truth), detect staleness lazily per the O7 lean, and +rebuild the root MOC explicitly. No LISTs on any hot path (D10) — the one +sanctioned walk lives in :func:`refresh_root_coverage`. + +Lives in its own module rather than ``zagg.hive`` to keep that file inside +the ~1000-line module guidance (review finding, PR #208 round 3): ``hive`` +owns the write side and the raw accessors; this module owns consumption. +""" + +from __future__ import annotations + +import json +import logging +import warnings + +import numpy as np + +from zagg.hive import ( + COVERAGE_SPEC, + MANIFEST_NAME, + ROOT_COVERAGE_NAME, + _decimal_base, + _decimal_order, + _decimal_rank, + _is_base_component, + build_root_coverage, + read_commit, + read_coverage_bitmap, + read_manifest, + read_root_coverage, + root_coverage_words, +) +from zagg.store import open_object_store + +logger = logging.getLogger(__name__) + +#: Stores already warned about a stale root MOC in this process (O7: warn +#: once per stale episode, trust silently otherwise, never auto-walk). +#: Keyed on the slash-normalized root; reset by a successful +#: :func:`refresh_root_coverage` so a LATER episode warns again. +_stale_warned: set[str] = set() + + +def load_coverage(store_root: str, **store_kwargs) -> dict | None: + """The store-root coverage envelope, or ``None`` when unusable. + + The tolerant reader-facing counterpart of + :func:`zagg.hive.read_root_coverage` (which raises on garbage JSON): + a missing object, unparsable JSON, or an unknown spec/encoding all read + as absent — the reader degrades to the walk, never to a wrong answer + (D9) — with a debug log so the degradation is discoverable. Strict spec + gate, the same posture as :func:`zagg.hive.read_coverage`. + """ + try: + envelope = read_root_coverage(store_root, **store_kwargs) + except ValueError as e: + logger.debug(f"unparsable {ROOT_COVERAGE_NAME} at {store_root} ({e}); ignoring") + return None + if envelope is None: + return None + usable = ( + isinstance(envelope, dict) + and envelope.get("spec") == COVERAGE_SPEC + and envelope.get("encoding") == "ranges" + ) + if not usable: + logger.debug(f"{ROOT_COVERAGE_NAME} at {store_root} has an unknown spec/encoding; ignoring") + return None + return envelope + + +def root_coverage_and(envelope: dict, aoi) -> np.ndarray: + """Intersection of the root MOC with an AOI morton cover. + + ``aoi`` is any morton cover (mixed order allowed — shard-order or + cell-order words both work; mortie's ``moc_and`` resolves containment + across orders). Returns the compacted intersection; an empty array means + the AOI touches no covered shard. Expansion is O(covered shards) — the + scale note on :func:`zagg.hive.root_coverage_words` applies. + """ + from mortie import moc_and + + return moc_and(root_coverage_words(envelope), np.asarray(aoi, dtype=np.uint64)) + + +def box_and(coverage: dict, aoi) -> np.ndarray: + """Intersection of a LEAF envelope's tier-0 box with an AOI morton cover. + + ``coverage`` is the stamp payload from :func:`zagg.hive.read_coverage`. + One in-memory op on <= 4 members — the cheap AOI reject readers run on + the stamp GET they already make, before paying for the bitmap sidecar. + An empty result rejects the leaf outright (the box is a conservative + superset: false positives possible, false negatives impossible). + """ + from mortie import moc_and + + from zagg.grids.morton import morton_word + + members = [morton_word(s) for s in coverage["box"] if s is not None] + return moc_and(np.asarray(members, dtype=np.uint64), np.asarray(aoi, dtype=np.uint64)) + + +def bitmap_and(leaf_root: str, aoi, **store_kwargs) -> np.ndarray | None: + """Exact cell-level intersection via a leaf's bitmap sidecar. + + One opt-in sidecar GET (:func:`zagg.hive.read_coverage_bitmap`), then + ``moc_and`` against ``aoi``. ``None`` when the leaf carries no bitmap + (box-only phase-1 stamp, depth-0 config, debris, absent leaf) — the + caller falls back to the box verdict; a present-but-corrupt sidecar + raises (the decoder's posture). An empty array is a definitive miss: + the bitmap is exact, not conservative. + """ + occupied = read_coverage_bitmap(leaf_root, **store_kwargs) + if occupied is None: + return None + from mortie import moc_and + + return moc_and(occupied, np.asarray(aoi, dtype=np.uint64)) + + +def _ranges_contain(envelope: dict, decimal: str) -> bool: + """Whether the envelope's ranges list one shard id — rank space, O(ranges). + + No word expansion (the containment check runs on the reader's hot path, + unlike the union's :func:`~zagg.hive.root_coverage_words`). + """ + if _decimal_order(decimal) != int(envelope["order"]): + return False + base, rank = _decimal_base(decimal), _decimal_rank(decimal) + return any( + _decimal_base(lo) == base and _decimal_rank(lo) <= rank <= _decimal_rank(hi) + for lo, hi in envelope["ranges"] + if _decimal_base(hi) == _decimal_base(lo) + ) + + +def warn_if_stale(store_root: str, shard_key, envelope: dict | None) -> bool: + """O7 lazy staleness detection for one opened, commit-stamped leaf. + + Call when a reader holds POSITIVE evidence of a committed shard — it + opened the leaf and read the stamp — that the store's root MOC does not + list. The most common cause is BENIGN: a run still in progress (the root + MOC is written at end of run, while leaves stamp continuously — review + finding, PR #208 round 4); the pathological causes are a crashed run, + the concurrent-run union race, and out-of-band writes. Returns ``True`` + when stale, warning ONCE per store per stale episode with the regen + suggestion (the latch is slash-normalized and reset by a successful + :func:`refresh_root_coverage`); afterwards it stays silent — the hot + path trusts silently and never auto-walks (D10). ``envelope`` may be + ``None`` (no root MOC at all): that is absence, not staleness, and reads + ``False``. Containment is checked in rank space (O(ranges), no word + expansion); a malformed envelope counts as not-listing, i.e. stale. + """ + if envelope is None: + return False + from zagg.grids.morton import morton_decimal + + decimal = morton_decimal(shard_key) + try: + if _ranges_contain(envelope, decimal): + return False + except (KeyError, TypeError, ValueError): + pass # malformed envelope cannot vouch for the shard — stale + key = store_root.rstrip("/") + if key not in _stale_warned: + _stale_warned.add(key) + warnings.warn( + f"commit-stamped shard {decimal} is not listed by {store_root}/" + f"{ROOT_COVERAGE_NAME} — the root MOC lags the leaves. Usually benign: a " + f"run still in progress writes the root MOC only at end of run. If no run " + f"is active, the causes are a crashed run, the concurrent-run union race, " + f"or out-of-band writes; regenerate with " + f"zagg.coverage.refresh_root_coverage({store_root!r})", + stacklevel=2, + ) + return True + + +def refresh_root_coverage(store_root: str, **store_kwargs) -> dict | None: + """Rebuild the root MOC from a full tree walk — the explicit escape hatch. + + THE SANCTIONED ROBUSTNESS PATH, not the hot path: D10 forbids walking + per read, but the strongly-consistent delimiter-LIST walk is ground + truth (§2), so an explicit ``refresh`` recovers from crashed runs, the + union race, and garbage objects. One carve-out (review finding, PR #208 + round 4): a stamped leaf at a NON-manifest order — hand-copied data or + an old-config survivor of a partial clear — cannot be represented in a + fixed-order ranges MOC; it is SKIPPED with a logged warning (this round + does not support mixed-order stores) and the root MOC is rebuilt from + the conforming leaves, so the escape hatch itself never dies on it. One + LIST per digit node (root: ``{sign+base}`` children; below: ``[1-4]`` + digits; a ``*.zarr`` entry is a leaf at that node), collecting the + shards whose commit stamp is present — unstamped debris is excluded, + exactly as the D4 model demands. The fresh envelope carries + ``source: "refresh"`` and REPLACES the root object (no union: the walk + supersedes it). A successful refresh also re-arms the + :func:`warn_if_stale` once-per-episode latch for this store. Returns the + envelope written, or ``None`` — deleting any existing root object — when + no stamped leaf exists (absence is truthful, a stale cache is not, and + the ranges envelope has no empty form). + """ + import obstore + from obstore.exceptions import NotFoundError + + from zagg.grids.morton import morton_word + from zagg.store import open_store + + manifest = read_manifest(store_root, **store_kwargs) + if manifest is None: + raise ValueError(f"no {MANIFEST_NAME} at {store_root} — not a hive store root") + order = int(manifest["shard_order"]) + store = open_object_store(store_root, **store_kwargs) + root = store_root.rstrip("/") + keys = [] + stack = [""] + while stack: + prefix = stack.pop() + listing = obstore.list_with_delimiter(store, prefix or None) + for child in listing["common_prefixes"]: + rel = child.rstrip("/") + name = rel.split("/")[-1] + if name.endswith(".zarr"): + decimal = name.removesuffix(".zarr") + if read_commit(open_store(f"{root}/{rel}", **store_kwargs)) is None: + continue # unstamped debris (D4) + if _decimal_order(decimal) != order: + # A fixed-order ranges MOC cannot represent this leaf: + # either corruption (old-config data surviving a partial + # clear, a hand-copied leaf) or a mixed-order future this + # round does not support. Skip it — the escape hatch must + # not die on the store it exists to repair. + logger.warning( + f"refresh: skipping stamped leaf {decimal} at order " + f"{_decimal_order(decimal)} under a shard_order-{order} manifest " + f"(mixed-order stores are unsupported; clear or re-shard the " + f"foreign-order data) — it will NOT be listed in {ROOT_COVERAGE_NAME}" + ) + continue + keys.append(morton_word(decimal)) + continue + is_digit_node = ( + _is_base_component(name) if prefix == "" else len(name) == 1 and name in "1234" + ) + if is_digit_node: + stack.append(rel + "/") + _stale_warned.discard(root) + if not keys: + try: + obstore.delete(store, ROOT_COVERAGE_NAME) + except (FileNotFoundError, NotFoundError): + pass + return None + envelope = build_root_coverage(keys, order, source="refresh") + obstore.put(store, ROOT_COVERAGE_NAME, json.dumps(envelope, indent=1).encode()) + return envelope + + +__all__ = [ + "bitmap_and", + "box_and", + "load_coverage", + "refresh_root_coverage", + "root_coverage_and", + "warn_if_stale", +] diff --git a/src/zagg/grids/morton.py b/src/zagg/grids/morton.py index d464104..56c389c 100644 --- a/src/zagg/grids/morton.py +++ b/src/zagg/grids/morton.py @@ -115,6 +115,48 @@ def morton_word(label: str) -> int: return _decimal_to_word(str(label)) +def morton_box(values) -> np.ndarray: + """Tier-0 morton box: canonical <= 4-member MOC covering ``values`` (issue #200). + + The fixed-width tier of the coverage envelope (``docs/design/ + sparse_coverage.md`` §4): compact the occupied cells to their canonical MOC + via mortie's ``compress_moc`` (mixed order allowed; an occupied ancestor + absorbs its descendants, complete sibling quads merge), then — unless one + member already covers everything — split at the deepest common ancestor + (``common_ancestor``, the longest common decimal-string prefix) and + **tighten** each of its 2-4 intersecting children to the common ancestor of + the occupancy inside it (review finding, PR #208). The result is + deterministic/canonical and a conservative superset of the input — no + occupied cell escapes the box — but NOT always the globally minimal + <= 4-member MOC (optimizing member areas under a 4-member cap is a harder + search); this "DCA children, each tightened" construction is the definition + that freezes with the mortie-side tier-0 spec. + + Accepts anything :func:`morton_words` does; returns sorted packed ``uint64`` + words. Raises ``ValueError`` on empty input (and, via mortie, on a set + spanning HEALPix base cells — a shard's cells are one subtree by + construction, so the hive path never triggers it). + """ + from mortie import clip2order, common_ancestor, compress_moc + + words = morton_words(values) + if words.size == 0: + raise ValueError("morton_box requires at least one occupied cell") + occ = compress_moc(words) + if occ.size == 1: + return occ + # After compression every member is strictly deeper than the ancestor (a + # member at the ancestor's own order would contain the rest and compress + # to a single cell), so coarsening lands each on one of its 2-4 children; + # the per-child common_ancestor then drops each member to the deepest + # cell covering that child's share of the occupancy. + anc = morton_decimal(common_ancestor(occ)) + anc_order = len(anc) - (2 if anc.startswith("-") else 1) + children = clip2order(anc_order + 1, occ) + box = [common_ancestor(occ[children == child]) for child in np.unique(children)] + return np.sort(np.asarray(box, dtype=np.uint64)) + + def morton_to_arrow(values): """Export ``values`` as a typed ``arro3.core.Array`` (issue #135). @@ -164,6 +206,7 @@ def is_morton_arrow(col) -> bool: "MORTON_EXTENSION_NAME", "is_morton_array", "is_morton_arrow", + "morton_box", "morton_decimal", "morton_from_arrow", "morton_to_arrow", diff --git a/src/zagg/hive.py b/src/zagg/hive.py index de91d40..855d0e9 100644 --- a/src/zagg/hive.py +++ b/src/zagg/hive.py @@ -26,13 +26,30 @@ debris — incomplete, ignorable, safe to overwrite on retry. This is NOT consolidated metadata: one small PUT on an object that must exist anyway. -The coverage MOC (§4) and pyramid sweep (§7) are follow-on issues; the -manifest's ``pyramid`` block is declared-only in round one (D11/D12). +- **Coverage (§4, issue #200)**: the stamp carries a ``coverage`` payload — + tier 0 is the shard's morton box, the canonical <= 4-member cover of its + occupied cells (:func:`zagg.grids.morton.morton_box`), padded to exactly + four decimal-string slots with JSON-null sentinels. Zero extra requests + (it rides the stamp PUT) and debris semantics are inherited: a torn + worker's coverage never becomes visible. Exact cell-order occupancy is a + zstd-compressed bitmap SIDECAR inside the leaf (``coverage.moc`` — the O8 + resolution; the one recorded exception to the vanilla-v3 leaf: data reads + are unaffected, but member enumeration warns and skips it), written + before the stamp and pointed to from the envelope; attrs stay lean and the + extra GET is paid only by readers that pass the box test. The optional end-of-run + root ``coverage.moc`` (issue #200 phase 3, default-on for hive) is a + shard-order ranges MOC at the store root — the second root-only object, + written fail-open by the dispatcher (locally) or a fire-and-forget worker + invoke (Lambda), and a regenerable cache under D9. The pyramid sweep (§7) + is a follow-on; the manifest's ``pyramid`` block is declared-only in round + one (D11/D12). """ from __future__ import annotations import json +import logging +import warnings from datetime import datetime, timezone import numpy as np @@ -41,12 +58,32 @@ from zagg.store import open_object_store +logger = logging.getLogger(__name__) + #: Convention version recorded in the manifest and the commit stamp (D6). HIVE_SPEC = "morton-hive/1" #: Root manifest object name (the root-only exception to the node invariant). MANIFEST_NAME = "morton_hive.json" #: Root-group attrs key carrying the commit stamp (D4). COMMIT_ATTR = "morton_hive_commit" +#: Convention version of the stamp's coverage payload (§4 tier 0, issue #200). +COVERAGE_SPEC = "morton-moc/1" +#: Fixed slot count of the tier-0 morton box (2-4 members, null-padded). +COVERAGE_BOX_SLOTS = 4 +#: In-leaf occupancy-bitmap sidecar object name (issue #200 phase 2, O8) — +#: the one recorded exception to the "vanilla zarr v3 leaf" claim: a foreign +#: key inside ``{full_id}.zarr/`` that zarr readers ignore (data reads are +#: unaffected; ``members()``/``tree()`` emit a ``ZarrUserWarning`` and skip +#: it — review finding, PR #208 round 2). +COVERAGE_SIDECAR = "coverage.moc" +#: zstd level for the sidecar bitmap — fixed so identical occupancy produces +#: byte-identical sidecars across workers and backends. +_ZSTD_LEVEL = 3 +#: Store-ROOT coverage object name (issue #200 phase 3): the shard-order MOC +#: for the one-GET bootstrap — the second root-only exception to the node +#: invariant, next to the manifest. Same name as the in-leaf sidecar +#: (:data:`COVERAGE_SIDECAR`), different location and encoding. +ROOT_COVERAGE_NAME = "coverage.moc" def shard_leaf_path(store_root: str, shard_key) -> str: @@ -189,22 +226,235 @@ def read_manifest(store_root: str, **store_kwargs) -> dict | None: return _read_json(open_object_store(store_root, **store_kwargs), MANIFEST_NAME) -def stamp_commit(leaf_store, *, cells_with_data: int, granule_count: int) -> None: +def build_coverage(shard_key, occupied, cell_order: int, *, bitmap: bytes | None = None) -> dict: + """Coverage payload for one shard's commit stamp (§4, issue #200). + + ``occupied`` is the shard's occupied cell words (mixed order allowed — + the cells ``cells_with_data`` counts); the box is their canonical + <= 4-member cover (:func:`zagg.grids.morton.morton_box`). ``None``/empty + falls back to the trivial 1-member cover, the shard id itself — always a + valid ancestor of its own coverage. Members are serialized as decimal + morton strings (D1), padded to exactly :data:`COVERAGE_BOX_SLOTS` slots + with trailing ``None`` (JSON null) sentinels — the recorded pad lean. + ``cell_order`` records the order occupancy was measured at; ``source`` + the producer (``"worker"`` at the leaf tier — phase-3 root and + sweep-composed payloads record theirs). ``generated_at`` is DELIBERATELY + omitted at the leaf (review finding, PR #208): the payload rides the + commit stamp, whose ``written_at`` is the one clock and one writer; + root/ancestor carriers add their own timestamp fields under this same + spec (per-carrier-optional). + + ``bitmap`` (phase 2, the O8 resolution) is the encoded sidecar payload + from :func:`encode_coverage_bitmap`; when given the envelope grows the + ``encoding``/``sidecar`` pointer plus compressed/raw byte sizes. A + box-only envelope (``None``, the phase-1 shape) omits those keys — a + reader treats their absence as "box only". Raises ``ValueError`` if the + box escapes the shard's subtree (occupied cells from another shard are + an upstream bug, never stamped). + """ + from zagg.grids.morton import morton_box, morton_decimal + + shard = morton_decimal(shard_key) + if occupied is None or len(occupied) == 0: + labels = [shard] + else: + labels = [morton_decimal(w) for w in morton_box(occupied)] + if len(labels) > COVERAGE_BOX_SLOTS or any(not s.startswith(shard) for s in labels): + raise ValueError( + f"coverage box {labels} escapes shard {shard}'s subtree — occupied " + f"cells must be the shard's own (the shard id is always a valid " + f"trivial cover, so this is an upstream cell-assignment bug)" + ) + coverage = { + "spec": COVERAGE_SPEC, + "box": labels + [None] * (COVERAGE_BOX_SLOTS - len(labels)), + "cell_order": int(cell_order), + "source": "worker", + } + if bitmap is not None: + n_bits = 4 ** (int(cell_order) - _decimal_order(shard)) + coverage.update( + encoding="bitmap", + sidecar=COVERAGE_SIDECAR, + nbytes=len(bitmap), + raw_nbytes=-(-n_bits // 8), + ) + return coverage + + +def _decimal_order(decimal: str) -> int: + """HEALPix order of a D1 decimal id (one digit per level past the base).""" + return len(decimal) - (2 if decimal.startswith("-") else 1) + + +def _cell_ranks(shard: str, cells, cell_order: int) -> np.ndarray: + """Bit index of each cell in the shard-subtree bitmap (frozen convention). + + Bit ``i`` is the i-th cell of the shard subtree at ``cell_order`` in + ascending packed-word (Z-)order — equivalently the base-4 value of the + cell's D1 digit tail with digits ``1..4`` mapped to ``0..3``. Raises + ``ValueError`` for a cell outside the subtree or not at ``cell_order`` + (the bitmap is exact-order by construction; there is nothing conservative + to fall back to). + """ + from zagg.grids.morton import to_morton_array + + depth = int(cell_order) - _decimal_order(shard) + ranks = np.empty(len(cells), dtype=np.int64) + for i, dec in enumerate(to_morton_array(cells).decimal_repr()): + tail = dec[len(shard) :] + if not dec.startswith(shard) or len(tail) != depth: + raise ValueError( + f"cell {dec} is not an order-{cell_order} cell of shard {shard}; " + f"the coverage bitmap encodes exact cell-order occupancy only" + ) + rank = 0 + for ch in tail: + rank = rank * 4 + (int(ch) - 1) + ranks[i] = rank + return ranks + + +def encode_coverage_bitmap(shard_key, occupied, cell_order: int) -> bytes: + """zstd-compressed exact occupancy bitmap for one shard (issue #200 phase 2). + + The O8-resolved leaf encoding: a bit field over the shard subtree at + ``cell_order`` — ``4^(cell_order - shard_order)`` bits, bit ``i`` per the + :func:`_cell_ranks` convention (ascending packed-word order; base-4 digit + tail), packed MSB-first within each byte (``np.packbits``), zstd- + compressed at a fixed level. Raw size is deterministic + (``ceil(4^depth / 8)`` bytes) regardless of fragmentation — the property + that beat coarsen-to-fit ranges in the #202 item (6) measurement; the + bit-order convention freezes with the mortie-side spec. zstd rides + numcodecs, already in the tree via zarr's codec stack — no new + dependency. + """ + from numcodecs import Zstd + + from zagg.grids.morton import morton_decimal + + shard = morton_decimal(shard_key) + depth = int(cell_order) - _decimal_order(shard) + if depth <= 0: + raise ValueError(f"cell_order {cell_order} is not below shard {shard}'s order") + # Staging is one uint8 per BIT — 8x the raw bitmap (1 MB at the design + # point: order-9 shards, order-19 cells). It is bounded by the shard's + # cell count, which the worker already materializes for the leaf + # template, so no extra guard here; coarse-shard + deep-cell configs + # beyond that envelope are out of scope (review note, PR #208 round 2). + bits = np.zeros(4**depth, dtype=np.uint8) + bits[_cell_ranks(shard, occupied, cell_order)] = 1 + return bytes(Zstd(level=_ZSTD_LEVEL).encode(np.packbits(bits).tobytes())) + + +def decode_coverage_bitmap(payload: bytes, shard_key, cell_order: int) -> np.ndarray: + """Occupied cell words from a sidecar bitmap payload (issue #200 phase 2). + + The inverse of :func:`encode_coverage_bitmap`: returns the sorted packed + ``uint64`` cell words at ``cell_order`` whose bits are set — exact + occupancy, no over-coverage. Posture (review finding, PR #208 round 2): + a CORRUPT payload — zstd garbage, or a decompressed size that is not the + exact raw bitmap size for the depth — raises loudly rather than + zero-padding/truncating to a plausible partial cell set (a false + negative, the one thing D9 forbids; the exact truth is intact in the + leaf, so surfacing beats under-reporting). A MISSING sidecar degrades to + ``None`` in :func:`read_coverage_bitmap`. + """ + from numcodecs import Zstd + + from zagg.grids.morton import morton_decimal, morton_word + + shard = morton_decimal(shard_key) + depth = int(cell_order) - _decimal_order(shard) + raw = np.frombuffer(bytes(Zstd().decode(payload)), dtype=np.uint8) + expected = -(-(4**depth) // 8) + if raw.size != expected: + raise ValueError( + f"coverage sidecar decompressed to {raw.size} B; an order-{cell_order} bitmap " + f"for shard {shard} is exactly {expected} B — refusing to zero-pad or truncate " + f"(a partial cell set would be a false negative)" + ) + bits = np.unpackbits(raw, count=4**depth) + words = np.empty(int(bits.sum()), dtype=np.uint64) + for i, rank in enumerate(np.flatnonzero(bits)): + digits, rank = [], int(rank) + for _ in range(depth): + digits.append(str(rank % 4 + 1)) + rank //= 4 + words[i] = morton_word(shard + "".join(reversed(digits))) + return np.sort(words) + + +def write_coverage_sidecar(leaf_root: str, payload: bytes, **store_kwargs) -> None: + """PUT the occupancy bitmap sidecar into a leaf (issue #200 phase 2). + + One object at ``{leaf}/coverage.moc`` — the recorded exception to the + vanilla-v3 leaf, ignored by zarr readers (member enumeration warns and + skips it; data reads are unaffected). Written BEFORE the commit + stamp so the stamp stays the leaf's FINAL write (D4): in an unstamped + prefix the sidecar is debris like everything else, and the wholesale + retry re-template clears it. + """ + import obstore + + obstore.put(open_object_store(leaf_root, **store_kwargs), COVERAGE_SIDECAR, payload) + + +def read_coverage_bitmap(leaf_root: str, **store_kwargs) -> np.ndarray | None: + """A leaf's exact occupied cell words from its sidecar, or ``None``. + + Gates on the committed stamp's envelope (:func:`read_coverage`): no + stamp, a box-only phase-1 payload (no ``encoding``/``sidecar`` keys), an + unknown encoding, or a missing sidecar object all read ``None`` — the + box is then the only index and readers degrade per D9, never to wrong + answers. A PRESENT-but-corrupt sidecar raises instead (see + :func:`decode_coverage_bitmap` — degrading a corrupt payload would be + indistinguishable from healthy box-only coverage). The shard id comes from the leaf's ``{full_id}.zarr`` basename; + ``cell_order`` from the envelope. One GET, paid only by readers that + want cell-level filtering. + """ + import obstore + from obstore.exceptions import NotFoundError + + from zagg.grids.morton import morton_word + from zagg.store import open_store + + coverage = read_coverage(open_store(leaf_root, **store_kwargs)) + if not coverage or coverage.get("encoding") != "bitmap" or not coverage.get("sidecar"): + return None + leaf_name = leaf_root.rstrip("/").rsplit("/", 1)[-1] + shard = morton_word(leaf_name.removesuffix(".zarr")) + store = open_object_store(leaf_root, **store_kwargs) + try: + data = obstore.get(store, str(coverage["sidecar"])).bytes() + except (FileNotFoundError, NotFoundError): + return None + return decode_coverage_bitmap(bytes(data), shard, int(coverage["cell_order"])) + + +def stamp_commit( + leaf_store, *, cells_with_data: int, granule_count: int, coverage: dict | None = None +) -> None: """Stamp a shard leaf complete — the shard's FINAL write (D4). One small PUT rewriting the leaf's root ``zarr.json`` (which the template already created), not consolidation. Until this lands, the leaf prefix is debris: a worker that dies mid-shard leaves no stamp, and a retry may - overwrite the prefix wholesale. + overwrite the prefix wholesale. ``coverage`` (issue #200) attaches the + tier-0 payload from :func:`build_coverage`; ``None`` writes the + pre-coverage stamp unchanged. """ group = zarr.open_group(leaf_store, path="", mode="r+", zarr_format=3) - group.attrs[COMMIT_ATTR] = { + stamp: dict = { "spec": HIVE_SPEC, "complete": True, "cells_with_data": int(cells_with_data), "granule_count": int(granule_count), "written_at": _utcnow(), } + if coverage is not None: + stamp["coverage"] = coverage + group.attrs[COMMIT_ATTR] = stamp def read_commit(leaf_store) -> dict | None: @@ -223,6 +473,187 @@ def read_commit(leaf_store) -> dict | None: return dict(stamp) if isinstance(stamp, dict) else None +def read_coverage(leaf_store) -> dict | None: + """The leaf's tier-0 coverage payload, or ``None`` when absent (issue #200). + + Rides :func:`read_commit`: debris and absent leaves read ``None``, and so + does a committed pre-coverage stamp (issue #199 stores carry no + ``coverage`` key) — older stores keep reading fine. STRICT on the spec + (review finding, PR #208): only ``spec == "morton-moc/1"`` payloads are + returned; a malformed dict or an unknown/future spec reads as absent + rather than half-parsed, so a new envelope version must be adopted here + deliberately instead of leaking through to box consumers. Box members are + decimal morton strings; parse one back with + :func:`zagg.grids.morton.morton_word`. + """ + stamp = read_commit(leaf_store) + if stamp is None: + return None + coverage = stamp.get("coverage") + if not isinstance(coverage, dict) or coverage.get("spec") != COVERAGE_SPEC: + return None + return dict(coverage) + + +def _decimal_base(decimal: str) -> str: + """The ``{sign+base}`` component of a D1 decimal id.""" + return decimal[:2] if decimal.startswith("-") else decimal[:1] + + +def _decimal_rank(decimal: str) -> int: + """Base-4 value of a D1 digit tail (digits ``1..4`` -> ``0..3``).""" + rank = 0 + for ch in decimal[len(_decimal_base(decimal)) :]: + rank = rank * 4 + (int(ch) - 1) + return rank + + +def _rank_tail(rank: int, depth: int) -> str: + """Inverse of :func:`_decimal_rank`: the width-``depth`` digit tail.""" + digits = [] + for _ in range(depth): + digits.append(str(rank % 4 + 1)) + rank //= 4 + return "".join(reversed(digits)) + + +def build_root_coverage(shard_keys, order: int, *, source: str = "dispatcher") -> dict: + """Store-root coverage envelope from completed shard keys (issue #200 phase 3). + + The O1 serialization: JSON ranges under the ``morton-moc/1`` envelope, + with ``encoding: "ranges"`` (vs the leaf sidecar's ``"bitmap"``), the + shard ``order``, ``source`` and ``generated_at`` — the root carrier's + staleness discriminators (per-carrier fields under the same spec; the + leaf payload deliberately omits them, see :func:`build_coverage`). A + range is an inclusive ``[first, last]`` run of same-order cells within + ONE base cell, consecutive in base-4 digit-tail rank (ascending + packed-word order — the bitmap's rank convention at the root). Endpoints + are D1 decimal STRINGS: packed u64 words exceed 2^53, so raw JSON + numbers would be silently mangled by any float-based parser (O1). + """ + from zagg.grids.morton import to_morton_array + + words = np.unique(np.asarray(shard_keys, dtype=np.uint64)) + if words.size == 0: + raise ValueError("build_root_coverage requires at least one shard key") + decs = list(to_morton_array(words).decimal_repr()) + bad = [d for d in decs if _decimal_order(d) != int(order)] + if bad: + raise ValueError(f"shard keys {bad[:3]} are not at shard order {order}") + # np.unique sorts by packed word; at a fixed order the words of one base + # cell are contiguous and rank-ascending, so one linear pass finds runs. + ranges = [] + start = prev = decs[0] + for dec in decs[1:]: + same_run = ( + _decimal_base(dec) == _decimal_base(prev) + and _decimal_rank(dec) == _decimal_rank(prev) + 1 + ) + if same_run: + prev = dec + continue + ranges.append([start, prev]) + start = prev = dec + ranges.append([start, prev]) + return { + "spec": COVERAGE_SPEC, + "encoding": "ranges", + "order": int(order), + "source": source, + "generated_at": _utcnow(), + "ranges": ranges, + } + + +def root_coverage_words(envelope: dict) -> np.ndarray: + """Shard words from a root envelope's ranges (inverse of the builder). + + Raises ``ValueError`` on malformed ranges (base-crossing, wrong order, + reversed endpoints) — same loud posture as the bitmap decoder: a corrupt + cache must never yield a plausible partial answer. + + Scale note (review, PR #208 round 3): expansion is O(covered shards) in + a Python loop — milliseconds at coherent-run scale (the design point, + shard order <= 11 regional products), but a full-sphere accumulated root + (~3M order-9 / ~50M order-11 shards) would take minutes worker-side. An + interval-space union on ``[base, lo_rank, hi_rank]`` triples (O(ranges), + no word materialization) is the upgrade path if root objects ever reach + continental-accumulation scale; out of scope here. + """ + from zagg.grids.morton import morton_word + + order = int(envelope["order"]) + words = [] + for lo, hi in envelope["ranges"]: + base = _decimal_base(lo) + lo_rank, hi_rank = _decimal_rank(lo), _decimal_rank(hi) + ok = _decimal_base(hi) == base and lo_rank <= hi_rank + ok = ok and _decimal_order(lo) == order and _decimal_order(hi) == order + if not ok: + raise ValueError(f"malformed coverage range [{lo}, {hi}] at order {order}") + words.extend(morton_word(base + _rank_tail(r, order)) for r in range(lo_rank, hi_rank + 1)) + return np.unique(np.asarray(words, dtype=np.uint64)) + + +def write_root_coverage(store_root: str, envelope: dict, **store_kwargs) -> dict: + """GET-union-PUT the store-root ``coverage.moc`` (issue #200 phase 3). + + Incremental runs accumulate: a parsable existing object with the same + spec/encoding/order is UNIONED with ``envelope`` before the PUT. An + unparsable or incompatible existing object is logged and OVERWRITTEN — + the root MOC is a regenerable cache (D9): the leaf stamps are the + durable truth and the §7 sweep is the authoritative rebuilder, so + merging with garbage would be worse than replacing it. CONCURRENT runs + race benignly (review finding, PR #208 round 3): GET-union-PUT is not + atomic and S3 has no compare-and-swap, so the last writer wins and its + union may miss the loser's shards until the sweep or the next run + re-unions — accepted under D9/O7 (a missing listing degrades to "reader + doesn't see the newest run", never a wrong answer; do NOT add a lock). + Returns the payload actually written. + """ + import obstore + + store = open_object_store(store_root, **store_kwargs) + try: + existing = _read_json(store, ROOT_COVERAGE_NAME) + except ValueError: + logger.warning( + f"existing {ROOT_COVERAGE_NAME} at {store_root} is not JSON; overwriting " + f"(regenerable cache — the sweep is the authoritative rebuilder)" + ) + existing = None + merged = envelope + if isinstance(existing, dict): + compatible = ( + existing.get("spec") == envelope.get("spec") + and existing.get("encoding") == envelope.get("encoding") + and existing.get("order") == envelope.get("order") + ) + if compatible: + try: + union = np.union1d(root_coverage_words(existing), root_coverage_words(envelope)) + merged = build_root_coverage( + union, int(envelope["order"]), source=envelope.get("source", "dispatcher") + ) + except (KeyError, TypeError, ValueError) as e: + logger.warning( + f"existing {ROOT_COVERAGE_NAME} at {store_root} failed to parse ({e}); " + f"overwriting (regenerable cache — the sweep rebuilds authoritatively)" + ) + else: + logger.warning( + f"existing {ROOT_COVERAGE_NAME} at {store_root} has an incompatible " + f"envelope; overwriting (regenerable cache)" + ) + obstore.put(store, ROOT_COVERAGE_NAME, json.dumps(merged, indent=1).encode()) + return merged + + +def read_root_coverage(store_root: str, **store_kwargs) -> dict | None: + """Read the store-root ``coverage.moc``; ``None`` when absent.""" + return _read_json(open_object_store(store_root, **store_kwargs), ROOT_COVERAGE_NAME) + + def leaf_block_index(grid, block_index, shard_key) -> tuple: """Leaf-LOCAL storage block for a chunk in a hive leaf (issue #199 phase 2). @@ -279,8 +710,14 @@ def _leaf(): store = open_store(leaf_path, **store_kwargs) # overwrite=True: any existing prefix here is either debris from a # torn run (D4) or a prior committed write being redone — both are - # replaced wholesale; per-leaf state never blocks a retry. - grid.emit_shard_template(store, overwrite=True) + # replaced wholesale; per-leaf state never blocks a retry. The + # overwrite enumeration warns about the prior attempt's coverage + # sidecar — the ONE foreign key we put there ourselves — so that + # specific warning is expected and suppressed; anything else in + # the prefix stays loud (review finding, PR #208 round 2). + with warnings.catch_warnings(): + warnings.filterwarnings("ignore", message=f"Object at {COVERAGE_SIDECAR}") + grid.emit_shard_template(store, overwrite=True) box["store"] = store return box["store"] @@ -296,6 +733,9 @@ def _write_chunk(block_index, carrier, ragged): ragged_key = shard_label(grid, shard_key) if single_chunk else int(local[0]) write_ragged_to_zarr(ragged, store, grid=grid, shard_key=ragged_key) + # Occupied-cell sink (issue #200): the worker already holds the shard's + # populated cell words; collect them here to derive the stamp's coverage. + occupied: list = [] _df_out, metadata = process_shard( grid, int(shard_key), @@ -306,16 +746,33 @@ def _write_chunk(block_index, carrier, ragged): handoff=handoff, aoi_payload=aoi_payload, write_chunk=_write_chunk, + occupied_out=occupied, profile=profile, ) # Stamp ONLY a fully-written leaf: an errored shard (or one that streamed # no chunks) stays unstamped — debris by definition (D4). The stamp is the - # last write, so its presence certifies everything before it landed. + # last write, so its presence certifies everything before it landed — the + # box payload rides it (zero extra requests), the exact-occupancy bitmap + # sidecar is PUT just before it (issue #200 phase 2), and both inherit + # its debris semantics: a torn worker's coverage never becomes visible. if "store" in box and not metadata.get("error"): + words = np.concatenate(occupied) if occupied else None + if words is not None and words.size == 0: + words = None + bitmap = None + # Depth 0 (child_order == parent_order, a legal one-cell-per-shard + # config) skips the sidecar: a 1-bit bitmap says nothing the stamp + # itself doesn't, and encode would raise AFTER the chunk writes, + # leaving the shard permanently unstampable debris (review finding, + # PR #208 round 2). The envelope simply omits the pointer — box only. + if words is not None and int(grid.child_order) > int(grid.parent_order): + bitmap = encode_coverage_bitmap(shard_key, words, grid.child_order) + write_coverage_sidecar(leaf_path, bitmap, **store_kwargs) stamp_commit( box["store"], cells_with_data=metadata.get("cells_with_data", 0), granule_count=metadata.get("granule_count", 0), + coverage=build_coverage(shard_key, words, grid.child_order, bitmap=bitmap), ) return metadata @@ -338,15 +795,29 @@ def _utcnow() -> str: __all__ = [ "COMMIT_ATTR", + "COVERAGE_BOX_SLOTS", + "COVERAGE_SIDECAR", + "COVERAGE_SPEC", "HIVE_SPEC", "MANIFEST_NAME", + "ROOT_COVERAGE_NAME", + "build_coverage", "build_manifest", + "build_root_coverage", "check_node_invariant", + "decode_coverage_bitmap", + "encode_coverage_bitmap", "ensure_manifest", "leaf_block_index", "process_and_write_hive", "read_commit", + "read_coverage", + "read_coverage_bitmap", "read_manifest", + "read_root_coverage", + "root_coverage_words", "shard_leaf_path", "stamp_commit", + "write_coverage_sidecar", + "write_root_coverage", ] diff --git a/src/zagg/processing/worker.py b/src/zagg/processing/worker.py index 3957c23..708c476 100644 --- a/src/zagg/processing/worker.py +++ b/src/zagg/processing/worker.py @@ -91,6 +91,7 @@ def process_shard( chunk_results: list | None = None, aoi_payload=None, write_chunk: Callable | None = None, + occupied_out: list | None = None, profile: bool = False, ) -> Tuple[pd.DataFrame, ProcessingMetadata]: """Process one shard: read granules, filter to this shard, aggregate, return df. @@ -176,6 +177,13 @@ def process_shard( the ``chunk_results`` / ``ragged_out`` behavior above is unchanged. The sharded path (#108) still bundles all K via ``chunk_results`` / ``write_shard_to_zarr`` and does not pass a callback. + occupied_out : list, optional + Out-param sink for the shard's occupied cells (issue #200). When a list + is passed, one ``uint64`` array of the distinct cell-order morton words + holding >= 1 observation — the cells ``cells_with_data`` counts — is + appended after the shard's reads are grouped. The hive write path uses + it to derive the commit stamp's coverage payload; ``None`` (default) + records nothing — byte-for-byte unchanged. profile : bool, optional Opt-in per-phase timing (issue #100 phase 2). When ``True``, fills ``metadata["phase_timings"]`` with ``read`` / ``index`` / ``aggregate`` @@ -529,6 +537,13 @@ def _iter_granule_reads(): col_arrays, cell_to_slice, n_obs_total = _concat_and_group(all_reads, grid, handoff) logger.info(f" Read {n_obs_total:,} observations") + # Occupied-cell sink (issue #200): both paths already key per-cell state by + # the packed cell word — ``cell_to_slice`` pooled, ``buffered.counts`` + # merged — so the occupied set is in hand with no extra observation pass. + if occupied_out is not None: + cells = buffered.counts if buffered is not None else cell_to_slice + occupied_out.append(np.fromiter(cells.keys(), dtype=np.uint64, count=len(cells))) + if profile: phase_timings["index"] = time.time() - _index_t0 _aggregate_t0 = time.time() diff --git a/src/zagg/runner.py b/src/zagg/runner.py index c7e7a82..987f13a 100644 --- a/src/zagg/runner.py +++ b/src/zagg/runner.py @@ -35,6 +35,7 @@ PipelineConfig, get_child_order, get_consolidate_metadata, + get_coverage_moc, get_driver, get_handoff, get_layout, @@ -1328,6 +1329,25 @@ def _accumulate(report, i, outcome): consolidate_metadata(zarr_store, zarr_format=3) wall_time = time.time() - start_time + # End-of-run root coverage.moc (issue #200 phase 3; default-on for hive, + # O9). Built from THIS run's successful completions and GET-unioned with + # any existing root object; the local dispatcher can write the store + # directly. Fail-open: the root MOC is a regenerable cache (D9) — a + # failed write costs readers one walk, never a wrong answer. + if store_layout == "hive" and get_coverage_moc(config): + from zagg.hive import build_root_coverage, write_root_coverage + + try: + # Inside the try so the fail-open claim survives result-envelope + # refactors (review finding, PR #208 round 3). + done = [m["shard_key"] for m in report.results if not m.get("error")] + if done: + envelope = build_root_coverage(done, int(grid.parent_order)) + write_root_coverage(store_path, envelope, **store_kwargs) + logger.info(f"Wrote root coverage.moc ({len(envelope['ranges'])} ranges)") + except Exception as e: + logger.warning(f"root coverage.moc write failed (fail-open, D9): {e}") + summary = { "total_cells": len(cells), "cells_with_data": report.cells_with_data, @@ -1661,6 +1681,61 @@ def _accumulate(report, i, result): finalize_s = time.time() - finalize_start wall_time = time.time() - start_time + # End-of-run root coverage.moc (issue #200 phase 3; default-on for hive, + # O9): the dispatcher cannot PUT to S3, so it builds + serializes the MOC + # and posts ONE fire-and-forget worker invoke that GET-unions-PUTs it. + # Transport rationale (espg-requested, plan question 3) — serialized + # ranges IN the event vs the completion list via the status channel: + # - Ranges (chosen): the dispatcher already holds the completion list + # in memory, so building the MOC costs milliseconds, and the payload + # is bounded by construction — spatially coherent coverage collapses + # to a few-KB range list, far under Lambda's 256 KB async-invoke cap, + # which a raw ~50k-key completion list would break. One hop, and no + # read-back race against status objects still landing from retried + # stragglers. + # - Completion list via .status/ (rejected): payload size would be + # run-independent and the artifact replayable from durable state, but + # it costs the worker a LIST + N GETs, races in-flight status writes, + # and its replayability is already owned by the §7 sweep's + # authoritative rebuild — the leaves are the durable truth (D9). + # Fail-open everywhere: a failed build/invoke logs and the run result is + # untouched (the root MOC is a regenerable cache). + if get_store_layout(config) == "hive" and get_coverage_moc(config): + try: + from zagg.hive import build_root_coverage + + # Inside the try so the fail-open claim survives result-envelope + # refactors (review finding, PR #208 round 3). + done = [ + r["shard_key"] + for r in report.results + if r.get("status_code") == 200 and not r.get("error") + ] + if done: + envelope = build_root_coverage(done, int(parent_order)) + # An OLD deployment has no coverage mode: the event falls + # through to its process handler, which returns a LOGGED 400 + # (missing shard_key/granule_urls...) — no writes, no result + # mirror, and no async redelivery (a returned 400 is a + # successful invocation to Lambda's Event retry machinery). + # Harmless under D9, but the CloudWatch line is an ERROR, not + # silence — mirroring the PR #205 deploy-ordering note. + logger.info( + f"Dispatching root coverage.moc write ({len(envelope['ranges'])} " + f"ranges, fire-and-forget) — requires a redeployed function; an " + f"older deployment 400s mode=coverage in its process handler " + f"(logged, no writes, no retry — harmless under D9)" + ) + _invoke_lambda_coverage( + state["lambda_client"], + function_name, + store_path, + envelope, + output_creds_event=output_creds_event, + ) + except Exception as e: + logger.warning(f"root coverage.moc dispatch failed (fail-open, D9): {e}") + # Cost estimate: arm64 pricing = $0.0000133334/GB-second. Compute gb_seconds # and cost *once* over the summed Lambda time (the report carries only the # accumulated compute_time_s) so the arithmetic order -- and thus the last @@ -2382,6 +2457,28 @@ def _invoke_lambda_finalize(lambda_client, function_name, store_path, output_cre raise RuntimeError(f"Lambda finalize error: {result.get('body')}") +def _invoke_lambda_coverage( + lambda_client, function_name, store_path, envelope, output_creds_event=None +): + """Fire-and-forget root ``coverage.moc`` write (issue #200 phase 3). + + ``InvocationType="Event"``: ~10 ms of dispatcher wall clock, run-size + independent, nothing blocks on it and no response is read — failure is + harmless by design (the worker-side GET-union-PUT is + ``zagg.hive.write_root_coverage``; the root MOC is a regenerable cache, + D9). The envelope rides pre-serialized in the event — see the dispatch + site for the transport rationale vs the status channel. + """ + event = {"mode": "coverage", "store_path": store_path, "coverage": envelope} + if output_creds_event is not None: + event["output_credentials"] = output_creds_event + lambda_client.invoke( + FunctionName=function_name, + InvocationType="Event", + Payload=json.dumps(event), + ) + + def _invoke_lambda_cell( lambda_client, chunk_idx, diff --git a/tests/test_coverage.py b/tests/test_coverage.py new file mode 100644 index 0000000..41eee29 --- /dev/null +++ b/tests/test_coverage.py @@ -0,0 +1,646 @@ +"""Hive coverage — issue #200 phases 1-3. + +Design contract: ``docs/design/sparse_coverage.md`` §4 (tiered coverage, as +amended on PR #206) plus the O8 resolution. Tier 0 is the morton box — the +canonical <= 4-member cover of a shard's occupied cells (DCA children, each +tightened — PR #208 finding 1), serialized as decimal strings padded to +exactly four JSON-null-sentinel slots, riding the D4 commit stamp with zero +extra store operations and inherited debris semantics. Exact occupancy is a +zstd-compressed bitmap SIDECAR inside the leaf (``coverage.moc``), written +before the stamp and pointed to from the envelope. Flat-layout stores are +untouched. The store-root ``coverage.moc`` (phase 3) is covered in +``tests/test_coverage_root.py`` (split at the phase-3 seam — review finding, +PR #208 round 3). +""" + +import importlib.util +import json +import os.path +from dataclasses import asdict +from pathlib import Path +from unittest.mock import MagicMock + +import numpy as np +import pandas as pd +import pytest +from zarr.storage import MemoryStore + +from zagg import hive +from zagg.config import default_config, get_data_vars +from zagg.grids import HealpixGrid +from zagg.grids.morton import morton_box, morton_decimal, morton_word + +# Order-6 southern shard used across the hive tests (decimal form -5112333). +SHARD = "-5112333" +# Its northern (positive-base) mirror: the ancestor-order arithmetic and the +# subtree prefix guard are string-form-dependent (no leading "-"), so the box +# matrix runs on both hemispheres (PR #208 finding 2). +NORTH = SHARD[1:] + + +def _words(*decimals): + return np.asarray([morton_word(d) for d in decimals], dtype=np.uint64) + + +def _decimals(words): + return [morton_decimal(w) for w in np.asarray(words, dtype=np.uint64)] + + +def _brute_force_box(decimals): + """Spec-literal oracle (issue #200 plan + PR #208 finding 1): the deepest + common ancestor is the longest common decimal-string prefix (D1: one digit + per level); the box is its intersecting children, each TIGHTENED to the + common prefix of the occupancy inside it — where an occupied cell EQUAL to + the ancestor occupies all of it. Pure string arithmetic, independent of + mortie's MOC kernels.""" + unique = sorted(set(decimals)) + if len(unique) == 1: + return unique + prefix = os.path.commonprefix(unique) + if prefix in unique: + return [prefix] + groups: dict = {} + for s in unique: + groups.setdefault(s[: len(prefix) + 1], []).append(s) + return sorted(os.path.commonprefix(g) for g in groups.values()) + + +def _canonical(words): + """Canonical compact form for area-equality comparison: two MOCs cover the + same area iff their compressed forms are identical (complete sibling quads + and their parent are the same area).""" + from mortie import compress_moc + + return compress_moc(np.asarray(words, dtype=np.uint64)) + + +def _random_occupancy(rng, shard, max_cells=24, max_depth=4): + """Random occupied cells at mixed depths within the ``shard`` subtree.""" + n = int(rng.integers(1, max_cells)) + return [ + shard + "".join(rng.choice(list("1234"), size=int(rng.integers(1, max_depth + 1)))) + for _ in range(n) + ] + + +# ── the box function ───────────────────────────────────────────────────────── + + +@pytest.fixture(params=[SHARD, NORTH], ids=["south", "north"]) +def shard(request): + return request.param + + +class TestMortonBox: + def test_single_cell_is_the_box(self, shard): + assert _decimals(morton_box(_words(shard + "1"))) == [shard + "1"] + + def test_duplicates_collapse(self, shard): + assert _decimals(morton_box(_words(shard + "12", shard + "12"))) == [shard + "12"] + + def test_two_cells_in_different_children(self, shard): + # DCA is the shard; each intersecting child is tightened to the lone + # occupied cell inside it (PR #208 finding 1). + box = morton_box(_words(shard + "12", shard + "43")) + assert _decimals(box) == [shard + "12", shard + "43"] + + def test_members_tightened_within_children(self, shard): + # The review counterexample: {S+111, S+112, S+2} must yield + # [S+11, S+2] — not the looser DCA-child form [S+1, S+2]. + box = morton_box(_words(shard + "111", shard + "112", shard + "2")) + assert _decimals(box) == [shard + "11", shard + "2"] + + def test_cells_spanning_all_four_children(self, shard): + box = morton_box(_words(*(shard + d + "1" for d in "1234"))) + assert _decimals(box) == [shard + d + "1" for d in "1234"] + + def test_mixed_depth_ancestor_absorbs_descendants(self, shard): + # An occupied ancestor covers its whole subtree: {parent, child} is + # the parent alone. (A naive "children of the DCA holding an occupied + # cell" would keep only the child and DROP the parent's remaining + # area — the superset test below is what pins this.) + box = morton_box(_words(shard + "1", shard + "14")) + assert _decimals(box) == [shard + "1"] + + def test_mixed_depth_split(self, shard): + box = morton_box(_words(shard + "11", shard + "422")) + assert _decimals(box) == [shard + "11", shard + "422"] + + def test_complete_quad_collapses_to_parent(self, shard): + # Four complete siblings tile their parent exactly; the canonical box + # is the parent — one member, area-identical to the four children. + box = morton_box(_words(*(shard + "1" + d for d in "1234"))) + assert _decimals(box) == [shard + "1"] + + def test_empty_raises(self): + with pytest.raises(ValueError, match="at least one"): + morton_box(np.asarray([], dtype=np.uint64)) + + @pytest.mark.parametrize("seed", range(8)) + def test_matches_brute_force_on_random_occupancy(self, shard, seed): + # Area-equal to the spec-literal tightened oracle, never > 4 members. + decs = _random_occupancy(np.random.default_rng(seed), shard) + box = morton_box(_words(*decs)) + assert 1 <= box.size <= hive.COVERAGE_BOX_SLOTS + np.testing.assert_array_equal(_canonical(box), _canonical(_words(*_brute_force_box(decs)))) + + @pytest.mark.parametrize("seed", range(8)) + def test_superset_invariant_randomized(self, shard, seed): + # Every occupied cell has a box member as ancestor (decimal prefix): + # false positives cost a wasted read, false negatives are impossible. + decs = _random_occupancy(np.random.default_rng(seed), shard) + box = _decimals(morton_box(_words(*decs))) + for d in decs: + assert any(d.startswith(b) for b in box), (d, box) + # And the box never escapes the shard subtree (the shard id is always + # a valid trivial ancestor). + assert all(b.startswith(shard) for b in box) + + +# ── the stamp payload ──────────────────────────────────────────────────────── + + +class TestBuildCoverage: + def _cov(self, *decimals, cell_order=8): + occupied = _words(*decimals) if decimals else None + return hive.build_coverage(morton_word(SHARD), occupied, cell_order) + + def test_exactly_four_slots_nulls_trail(self): + for occ, members in [ + ((SHARD + "11",), 1), + ((SHARD + "12", SHARD + "43"), 2), + ((SHARD + "11", SHARD + "21", SHARD + "31"), 3), + (tuple(SHARD + d + "1" for d in "1234"), 4), + ]: + cov = self._cov(*occ) + assert len(cov["box"]) == hive.COVERAGE_BOX_SLOTS + assert all(isinstance(s, str) for s in cov["box"][:members]) + assert cov["box"][members:] == [None] * (hive.COVERAGE_BOX_SLOTS - members) + + def test_payload_fields(self): + cov = self._cov(SHARD + "12", SHARD + "43", cell_order=8) + assert cov == { + "spec": "morton-moc/1", + "box": [SHARD + "12", SHARD + "43", None, None], + "cell_order": 8, + "source": "worker", + } + # JSON-safe as-is: the null sentinel is the recorded pad lean. + assert json.loads(json.dumps(cov)) == cov + + def test_box_members_round_trip_decimal(self): + cov = self._cov(SHARD + "12", SHARD + "43") + for label in cov["box"]: + if label is not None: + assert morton_decimal(morton_word(label)) == label + + def test_empty_occupancy_falls_back_to_trivial_shard_cover(self): + # The shard id is always a valid (trivial) 1-member cover. + assert self._cov()["box"] == [SHARD, None, None, None] + + def test_whole_shard_occupancy_is_the_shard_itself(self): + cov = self._cov(*(SHARD + d for d in "1234")) + assert cov["box"] == [SHARD, None, None, None] + + def test_occupancy_outside_shard_subtree_rejected(self): + with pytest.raises(ValueError, match="subtree"): + self._cov(SHARD[:-1] + "41") # sibling shard's cell + + def test_northern_shard_subtree_check(self): + # The prefix guard has no sign character on positive bases (PR #208 + # finding 2): both the accept and the reject arms run northern. + cov = hive.build_coverage(morton_word(NORTH), _words(NORTH + "12"), 8) + assert cov["box"] == [NORTH + "12", None, None, None] + with pytest.raises(ValueError, match="subtree"): + hive.build_coverage(morton_word(NORTH), _words(NORTH[:-1] + "41"), 8) + + +class TestStampCoverage: + def _stamped_store(self, coverage): + cfg = default_config("atl06") + store = MemoryStore() + grid = HealpixGrid(parent_order=6, child_order=8, layout="fullsphere", config=cfg) + grid.emit_shard_template(store, overwrite=True) + hive.stamp_commit(store, cells_with_data=2, granule_count=1, coverage=coverage) + return store + + def test_stamp_carries_coverage_and_accessor_reads_it(self): + cov = hive.build_coverage(morton_word(SHARD), _words(SHARD + "12", SHARD + "43"), 8) + store = self._stamped_store(cov) + stamp = hive.read_commit(store) + assert stamp["coverage"] == cov + assert hive.read_coverage(store) == cov + # No timestamp of its own: the payload reuses the stamp's written_at. + assert "written_at" not in cov and "generated_at" not in cov + assert stamp["written_at"] + + def test_stamp_payload_byte_cost(self): + # The byte-cost pin promised in the plan (issue #200) and PR #208 + # finding 3: the tier-0 payload is fixed-width by construction — keep + # it (and future envelope creep) bounded on the leaf zarr.json every + # reader GETs. + cov = hive.build_coverage(morton_word(SHARD), _words(*(SHARD + d + "1" for d in "1234")), 8) + assert len(json.dumps(cov)) < 256 + + def test_pre_coverage_stamp_reads_none(self): + # Forward compat: an issue-#199 stamp (no coverage key) keeps reading + # fine — commit still visible, coverage reads None. + store = self._stamped_store(None) + assert hive.read_commit(store)["complete"] is True + assert hive.read_coverage(store) is None + + def test_unknown_spec_reads_none(self): + # Strict spec posture (PR #208 finding 4): an unknown/future envelope + # version reads as absent, never half-parsed. The raw stamp keeps the + # payload for whoever understands it. + cov = {"spec": "morton-moc/2", "box": [SHARD, None, None, None], "cell_order": 8} + store = self._stamped_store(cov) + assert hive.read_coverage(store) is None + assert hive.read_commit(store)["coverage"] == cov + + def test_missing_spec_reads_none(self): + store = self._stamped_store({"box": [SHARD, None, None, None]}) + assert hive.read_coverage(store) is None + + def test_debris_and_absent_leaves_read_none(self): + assert hive.read_coverage(MemoryStore()) is None # no leaf at all + cfg = default_config("atl06") + store = MemoryStore() + grid = HealpixGrid(parent_order=6, child_order=8, layout="fullsphere", config=cfg) + grid.emit_shard_template(store, overwrite=True) # unstamped = debris + assert hive.read_coverage(store) is None + + +# ── the occupancy bitmap (phase 2, O8) ─────────────────────────────────────── + + +class TestCoverageBitmap: + """Exact cell-order occupancy as a zstd bitmap: bit i = the i-th subtree + cell in ascending packed-word order (base-4 D1 digit tail, 1..4 -> 0..3), + MSB-first per byte — the frozen encoding convention.""" + + def _tails(self, rng, depth, n): + return {"".join(rng.choice(list("1234"), size=depth)) for _ in range(n)} + + def test_round_trip_exact(self, shard): + occ = _words(shard + "12", shard + "43", shard + "21") + payload = hive.encode_coverage_bitmap(morton_word(shard), occ, _order(shard) + 2) + decoded = hive.decode_coverage_bitmap(payload, morton_word(shard), _order(shard) + 2) + np.testing.assert_array_equal(decoded, np.sort(occ)) + + @pytest.mark.parametrize("seed", range(6)) + def test_round_trip_property(self, shard, seed): + # Exactness: the bitmap decodes to exactly the occupied set — no + # over-coverage, no false negatives (property over random occupancy). + rng = np.random.default_rng(seed) + depth = 3 + occ = _words(*(shard + t for t in self._tails(rng, depth, int(rng.integers(1, 40))))) + cell_order = _order(shard) + depth + decoded = hive.decode_coverage_bitmap( + hive.encode_coverage_bitmap(morton_word(shard), occ, cell_order), + morton_word(shard), + cell_order, + ) + np.testing.assert_array_equal(decoded, np.sort(occ)) + + def test_deterministic_bytes(self, shard): + # Same occupancy (any input order, duplicates included) -> the + # byte-identical sidecar: the backend-identity claim at byte level. + word, order = morton_word(shard), _order(shard) + 2 + a = hive.encode_coverage_bitmap(word, _words(shard + "12", shard + "43"), order) + b = hive.encode_coverage_bitmap( + word, _words(shard + "43", shard + "12", shard + "43"), order + ) + assert a == b + + def test_zstd_compresses_fragmented_occupancy(self): + # Realistic fragmentation (the #202 measurement's regime: scattered + # cells, ~1 cell per run): compressed strictly below the deterministic + # raw size. + rng = np.random.default_rng(0) + depth = 7 # 16384 bits = 2 KB raw + occ = _words(*(SHARD + t for t in self._tails(rng, depth, 800))) + payload = hive.encode_coverage_bitmap(morton_word(SHARD), occ, _order(SHARD) + depth) + assert len(payload) < 4**depth // 8 + + def test_golden_raw_bytes(self, shard): + # THE WIRE-FORMAT PIN (review finding, PR #208 round 2): round-trip + # and determinism tests pass under ANY self-consistent bit + # permutation, so only fixed raw-byte vectors freeze the convention + # (rank = ascending packed-word order, MSB-first per byte) against a + # silent flip. Decompressed bytes pinned — compressed bytes are zstd- + # library-version-dependent; raw bytes are the frozen convention. + from numcodecs import Zstd + + word, order = morton_word(shard), _order(shard) + 2 + # tail "11" = rank 0 -> MSB of byte 0; tail "44" = rank 15 -> LSB of byte 1 + one = bytes(Zstd().decode(hive.encode_coverage_bitmap(word, _words(shard + "11"), order))) + assert one == b"\x80\x00" + last = bytes(Zstd().decode(hive.encode_coverage_bitmap(word, _words(shard + "44"), order))) + assert last == b"\x00\x01" + # Depth-3 multi-cell vector freezes the rank arithmetic too: + # tails 111/114/241/444 -> ranks 0, 3, 28, 63. + occ = _words(*(shard + t for t in ("111", "114", "241", "444"))) + multi = bytes(Zstd().decode(hive.encode_coverage_bitmap(word, occ, _order(shard) + 3))) + assert multi == b"\x90\x00\x00\x08\x00\x00\x00\x01" + + def test_truncated_payload_rejected(self, shard): + # A wrong-sized payload must raise, never zero-pad to a partial cell + # set (false negatives, D9 — review finding, PR #208 round 2). The + # reviewer's case: 1 raw byte where depth 3 needs 8. + from numcodecs import Zstd + + short = bytes(Zstd(level=3).encode(b"\xff")) + with pytest.raises(ValueError, match="refusing to zero-pad"): + hive.decode_coverage_bitmap(short, morton_word(shard), _order(shard) + 3) + oversized = bytes(Zstd(level=3).encode(b"\xff" * 16)) + with pytest.raises(ValueError, match="refusing to zero-pad"): + hive.decode_coverage_bitmap(oversized, morton_word(shard), _order(shard) + 3) + + def test_wrong_order_cell_rejected(self): + with pytest.raises(ValueError, match="exact cell-order"): + hive.encode_coverage_bitmap(morton_word(SHARD), _words(SHARD + "1"), _order(SHARD) + 2) + + def test_cell_outside_shard_rejected(self): + with pytest.raises(ValueError, match="exact cell-order"): + hive.encode_coverage_bitmap( + morton_word(SHARD), _words(SHARD[:-1] + "412"), _order(SHARD) + 2 + ) + + +def _order(decimal): + return len(decimal) - (2 if decimal.startswith("-") else 1) + + +# ── the worker seam ────────────────────────────────────────────────────────── + + +class TestOccupiedOutSink: + """The real ``process_shard`` seam: ``occupied_out`` receives exactly the + distinct cell words holding observations (the cells ``cells_with_data`` + counts), on the same read stub the write-path tests use.""" + + def test_occupied_out_gets_cells_with_data(self, monkeypatch): + from zagg.index.hierarchical import HierarchicalIndex + from zagg.processing import process_shard + + cfg = default_config() + grid = HealpixGrid(6, 8, layout="fullsphere", config=cfg) + shard = morton_word(SHARD) + children = grid.children(shard) + c1, c2 = int(children[0]), int(children[5]) + df = pd.DataFrame( + { + "h_li": np.array([3.0, 1.0, 7.0], dtype=np.float32), + "s_li": np.array([0.1, 0.1, 0.1], dtype=np.float32), + "leaf_id": np.array([c1, c1, c2], dtype=np.uint64), + } + ) + calls = {"n": 0} + + def one_shot(*args, **kwargs): + calls["n"] += 1 + return df if calls["n"] == 1 else None + + monkeypatch.setattr("zagg.processing._read_group", one_shot) + monkeypatch.setattr("zagg.processing.h5coro.H5Coro", lambda *a, **k: object()) + monkeypatch.setattr("zagg.processing._make_url_rewriter", lambda driver: lambda u: u) + monkeypatch.setattr( + "zagg.processing.worker.index_from_config", lambda cfg: HierarchicalIndex() + ) + + occupied: list = [] + _df, meta = process_shard( + grid, + shard, + ["s3://x"], + s3_credentials={}, + config=cfg, + chunk_results=[], + occupied_out=occupied, + ) + (words,) = occupied + assert words.dtype == np.uint64 + assert sorted(int(w) for w in words) == sorted({c1, c2}) + assert meta["cells_with_data"] == 2 + # Phase 2 exactness, against the REAL worker seam's occupancy: the + # sidecar bitmap round-trips to exactly the occupied set. + payload = hive.encode_coverage_bitmap(shard, words, grid.child_order) + decoded = hive.decode_coverage_bitmap(payload, shard, grid.child_order) + assert sorted(int(w) for w in decoded) == sorted({c1, c2}) + + +# ── the hive write path (both backends) ────────────────────────────────────── + + +def _rec_meta(shard): + return { + "shard_key": int(shard), + "cells_with_data": 2, + "total_obs": 7, + "granule_count": 1, + "files_processed": 1, + "duration_s": 0.0, + "error": None, + } + + +def _carrier(grid, shard): + coords = grid.chunk_coords(shard) + n = len(coords["cell_ids"]) + df = pd.DataFrame( + { + var: np.zeros(n, dtype=np.int32 if var == "count" else np.float32) + for var in get_data_vars(grid.config) + } + ) + for name, vals in coords.items(): + df[name] = vals + return df + + +def _occupancy_fake(grid, occupied_words): + """A process_shard stand-in that streams one real carrier and reports + ``occupied_words`` through the occupied_out sink, as the real worker does.""" + + def fake(g, shard_key, urls, **kwargs): + kwargs["write_chunk"](grid.block_index(int(shard_key)), _carrier(grid, shard_key), {}) + if kwargs.get("occupied_out") is not None and occupied_words is not None: + kwargs["occupied_out"].append(np.asarray(occupied_words, dtype=np.uint64)) + return pd.DataFrame(), _rec_meta(shard_key) + + return fake + + +class TestProcessAndWriteHiveCoverage: + def _run(self, monkeypatch, tmp_path, occupied_words): + import zagg.processing as processing + from zagg.store import open_store + + cfg = default_config("atl06") + grid = HealpixGrid(parent_order=6, child_order=8, layout="fullsphere", config=cfg) + shard = morton_word(SHARD) + monkeypatch.setattr(processing, "process_shard", _occupancy_fake(grid, occupied_words)) + root = str(tmp_path / "store") + hive.process_and_write_hive(shard, ["s3://b/g1.h5"], grid, {}, root, cfg, store_kwargs={}) + leaf = hive.shard_leaf_path(root, shard) + return grid, leaf, open_store(leaf) + + def test_stamp_carries_envelope_and_sidecar_pointer(self, monkeypatch, tmp_path): + import os + + occupied = _words(SHARD + "12", SHARD + "43") + grid, leaf, leaf_store = self._run(monkeypatch, tmp_path, occupied) + sidecar = os.path.join(leaf, hive.COVERAGE_SIDECAR) + cov = hive.read_coverage(leaf_store) + assert cov == { + "spec": hive.COVERAGE_SPEC, + "box": [SHARD + "12", SHARD + "43", None, None], + "cell_order": int(grid.child_order), + "source": "worker", + "encoding": "bitmap", + "sidecar": hive.COVERAGE_SIDECAR, + "nbytes": os.path.getsize(sidecar), + "raw_nbytes": 4 ** (grid.child_order - grid.parent_order) // 8, + } + assert hive.read_commit(leaf_store)["coverage"] == cov + # Attrs stay lean: the envelope is bounded well under 1 KB — the + # exact payload lives in the sidecar, not the zarr.json readers GET. + assert len(json.dumps(cov)) < 512 + # The sidecar decodes to exactly the worker's occupied set. + np.testing.assert_array_equal(hive.read_coverage_bitmap(leaf), np.sort(occupied)) + + def test_worker_without_occupancy_stamps_box_only(self, monkeypatch, tmp_path): + # No occupied_out delivery (e.g. a legacy caller): the shard id is the + # trivial 1-member cover, no sidecar, and the envelope omits the + # encoding/pointer keys — the phase-1 box-only shape, which the + # bitmap reader treats as "box only" (forward/back compat). + import os + + _grid, leaf, leaf_store = self._run(monkeypatch, tmp_path, None) + cov = hive.read_coverage(leaf_store) + assert cov["box"] == [SHARD, None, None, None] + assert "encoding" not in cov and "sidecar" not in cov + assert not os.path.exists(os.path.join(leaf, hive.COVERAGE_SIDECAR)) + assert hive.read_coverage_bitmap(leaf) is None + + def test_depth_zero_config_stamps_box_only(self, monkeypatch, tmp_path): + # child_order == parent_order is a legal one-cell-per-shard config + # (review finding, PR #208 round 2): the sidecar is skipped — a 1-bit + # bitmap says nothing the stamp doesn't — and the shard still stamps + # a box-only envelope instead of dying unstampable after its writes. + import os + + import zagg.processing as processing + from zagg.store import open_store + + cfg = default_config("atl06") + grid = HealpixGrid(parent_order=6, child_order=6, layout="fullsphere", config=cfg) + word = morton_word(SHARD) + occupied = _words(SHARD) # the lone cell IS the shard at depth 0 + monkeypatch.setattr(processing, "process_shard", _occupancy_fake(grid, occupied)) + root = str(tmp_path / "store") + hive.process_and_write_hive(word, ["s3://b/g1.h5"], grid, {}, root, cfg, store_kwargs={}) + leaf = hive.shard_leaf_path(root, word) + cov = hive.read_coverage(open_store(leaf)) + assert cov["box"] == [SHARD, None, None, None] + assert "sidecar" not in cov and "encoding" not in cov + assert not os.path.exists(os.path.join(leaf, hive.COVERAGE_SIDECAR)) + assert hive.read_coverage_bitmap(leaf) is None + + def test_unstamped_sidecar_is_debris(self, tmp_path): + # A sidecar in an UNSTAMPED prefix is debris: the accessor gates on + # the committed stamp, so it never becomes visible (D4 semantics). + cfg = default_config("atl06") + grid = HealpixGrid(parent_order=6, child_order=8, layout="fullsphere", config=cfg) + from zagg.store import open_store + + leaf = hive.shard_leaf_path(str(tmp_path / "store"), morton_word(SHARD)) + grid.emit_shard_template(open_store(leaf), overwrite=True) # no stamp + payload = hive.encode_coverage_bitmap(morton_word(SHARD), _words(SHARD + "12"), 8) + hive.write_coverage_sidecar(leaf, payload) + assert hive.read_coverage_bitmap(leaf) is None + + +# ── backend identity via the Lambda handler path ───────────────────────────── + +HANDLER_PATH = Path(__file__).parent.parent / "deployment" / "aws" / "lambda_handler.py" + + +@pytest.fixture(scope="module") +def handler_mod(): + spec = importlib.util.spec_from_file_location("zagg_lambda_handler_coverage", HANDLER_PATH) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +class TestBothBackendsIdenticalPayload: + """The shared-code-path pin: the local dispatcher and the Lambda handler + both run ``hive.process_and_write_hive``, so identical input produces the + byte-identical coverage payload — one test through the REAL handler path, + like the existing stamp tests.""" + + # Order-6 southern shard the handler tests use (decimal -4211322). + _WORD = 11827859996358475782 + + def test_local_and_lambda_coverage_payloads_identical(self, handler_mod, monkeypatch, tmp_path): + import zagg.processing as processing + from zagg.config import load_config_from_dict + from zagg.grids import from_config + from zagg.store import open_store + + # Self-recycle hygiene (issue #171): never let this module-scoped + # handler instance reach a real os._exit under dev-shell env knobs. + monkeypatch.delenv("ZAGG_RECYCLE_RSS_MB", raising=False) + monkeypatch.delenv("ZAGG_RECYCLE_MAX_INVOCATIONS", raising=False) + monkeypatch.setattr( + handler_mod, "_exit", lambda code: (_ for _ in ()).throw(AssertionError(code)) + ) + + cfg = default_config("atl06") + cfg.output["store_layout"] = "hive" + config_dict = asdict(cfg) + grid = from_config(load_config_from_dict(config_dict)) + shard_dec = morton_decimal(self._WORD) + depth = int(grid.child_order) - _order(shard_dec) + occupied = _words(shard_dec + "1" * depth, shard_dec + "4" * depth) + monkeypatch.setattr(processing, "process_shard", _occupancy_fake(grid, occupied)) + + # Local backend leg. + local_root = str(tmp_path / "local") + hive.process_and_write_hive( + self._WORD, ["s3://b/g.h5"], grid, {}, local_root, cfg, store_kwargs={} + ) + + # Lambda backend leg: the real process-mode handler. + lambda_root = str(tmp_path / "lambda") + ctx = MagicMock() + ctx.aws_request_id = "req-1" + ctx.function_name = "process-shard" + ctx.memory_limit_in_mb = 2048 + ctx.get_remaining_time_in_millis.return_value = 900_000 + event = { + "shard_key": self._WORD, + "parent_order": 6, + "child_order": int(grid.child_order), + "granule_urls": ["s3://b/g.h5"], + "store_path": lambda_root, + "s3_credentials": {"accessKeyId": "a", "secretAccessKey": "s", "sessionToken": "t"}, + "config": config_dict, + } + resp = handler_mod._handle_process(event, ctx) + assert resp["statusCode"] == 200, resp["body"] + + local_leaf = hive.shard_leaf_path(local_root, self._WORD) + lambda_leaf = hive.shard_leaf_path(lambda_root, self._WORD) + local_cov = hive.read_coverage(open_store(local_leaf)) + lambda_cov = hive.read_coverage(open_store(lambda_leaf)) + assert local_cov is not None + assert local_cov == lambda_cov + assert local_cov["box"] == [shard_dec + "1" * depth, shard_dec + "4" * depth, None, None] + # And the sidecars are byte-identical: same occupancy, same encoding + # convention, same fixed zstd level on both backends. + local_bytes = Path(local_leaf, hive.COVERAGE_SIDECAR).read_bytes() + assert local_bytes == Path(lambda_leaf, hive.COVERAGE_SIDECAR).read_bytes() + np.testing.assert_array_equal(hive.read_coverage_bitmap(local_leaf), np.sort(occupied)) diff --git a/tests/test_coverage_root.py b/tests/test_coverage_root.py new file mode 100644 index 0000000..789c5d8 --- /dev/null +++ b/tests/test_coverage_root.py @@ -0,0 +1,715 @@ +"""Store-root ``coverage.moc`` — issue #200 phases 3-4. + +The end-of-run shard-order ranges MOC (O1 serialization, default-on for hive, +fail-open on both backends): the config flag, the envelope builder/parser, the +GET-union-PUT writer, the local and Lambda dispatcher legs, and the worker's +``mode: "coverage"`` handler — plus the phase-4 reader primitives in +``zagg.coverage`` (envelope load, per-tier AOI intersection, O7 lazy +staleness, the explicit refresh walk). Split from ``tests/test_coverage.py`` +at the phase-3 seam (review finding, PR #208 round 3); the leaf tiers (box, +bitmap, stamp envelope) live there. +""" + +import importlib.util +import json +import warnings +from pathlib import Path +from unittest.mock import MagicMock + +import numpy as np +import pytest + +from zagg import hive +from zagg.config import default_config +from zagg.grids.morton import morton_word + +# Order-6 southern shard shared with the leaf-tier suite (decimal -5112333). +SHARD = "-5112333" + +HANDLER_PATH = Path(__file__).parent.parent / "deployment" / "aws" / "lambda_handler.py" + + +@pytest.fixture(scope="module") +def handler_mod(): + spec = importlib.util.spec_from_file_location("zagg_lambda_handler_root_coverage", HANDLER_PATH) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def _words(*decimals): + return np.asarray([morton_word(d) for d in decimals], dtype=np.uint64) + + +# ── the root coverage MOC (phase 3) ────────────────────────────────────────── + + +class TestCoverageMocConfig: + """O9: default ON for hive (healpix-only by validation); explicit true + anywhere it cannot land is a pointed error, absent there is simply off.""" + + def _cfg(self, **output): + cfg = default_config("atl06") + cfg.output.update(output) + return cfg + + def test_default_on_for_hive(self): + from zagg.config import get_coverage_moc, validate_config + + cfg = self._cfg(store_layout="hive") + assert get_coverage_moc(cfg) is True + validate_config(cfg) + + def test_default_off_for_flat(self): + from zagg.config import get_coverage_moc, validate_config + + cfg = self._cfg() + assert get_coverage_moc(cfg) is False + validate_config(cfg) + + def test_explicit_off_for_hive(self): + from zagg.config import get_coverage_moc, validate_config + + cfg = self._cfg(store_layout="hive", coverage_moc=False) + assert get_coverage_moc(cfg) is False + validate_config(cfg) + + def test_null_falls_back_to_default(self): + from zagg.config import get_coverage_moc, validate_config + + cfg = self._cfg(store_layout="hive", coverage_moc=None) + assert get_coverage_moc(cfg) is True + validate_config(cfg) + + def test_explicit_true_on_flat_rejected(self): + from zagg.config import validate_config + + with pytest.raises(ValueError, match="store_layout: hive"): + validate_config(self._cfg(coverage_moc=True)) + + def test_explicit_true_on_rectilinear_rejected(self): + from zagg.config import validate_config + + cfg = self._cfg(coverage_moc=True) + cfg.output["grid"] = { + "type": "rectilinear", + "crs": "EPSG:3031", + "resolution": 100, + "bounds": [0, 0, 1000, 1000], + } + with pytest.raises(ValueError, match="healpix"): + validate_config(cfg) + + def test_non_bool_rejected(self): + from zagg.config import validate_config + + with pytest.raises(ValueError, match="coverage_moc must be a boolean"): + validate_config(self._cfg(store_layout="hive", coverage_moc="yes")) + + +class TestRootCoverage: + """The O1 envelope: shard-order ranges MOC, decimal-string endpoints.""" + + def test_envelope_fields_and_run_collapse(self): + keys = _words("-511", "-512", "-513", "-521", "511", "512") + env = hive.build_root_coverage(keys, 2) + assert env["spec"] == hive.COVERAGE_SPEC + assert env["encoding"] == "ranges" + assert env["order"] == 2 + assert env["source"] == "dispatcher" + assert env["generated_at"] + # Consecutive ranks collapse to one range; runs never span base cells. + assert ["-511", "-513"] in env["ranges"] + assert ["-521", "-521"] in env["ranges"] + assert ["511", "512"] in env["ranges"] + assert len(env["ranges"]) == 3 + # Endpoints are STRINGS (packed u64 words exceed 2^53 — the JSON + # float-parser trap O1 specs around), and the payload is JSON-safe. + assert all(isinstance(e, str) for pair in env["ranges"] for e in pair) + assert json.loads(json.dumps(env)) == env + + def test_words_round_trip_exact(self): + keys = _words("-511", "-512", "-514", "521", "611") + env = hive.build_root_coverage(keys, 2) + np.testing.assert_array_equal(hive.root_coverage_words(env), np.sort(keys)) + + @pytest.mark.parametrize("seed", range(4)) + def test_words_round_trip_property(self, seed): + rng = np.random.default_rng(seed) + bases = ["-5", "5", "-1", "3"] + decs = { + rng.choice(bases) + "".join(rng.choice(list("1234"), size=4)) + for _ in range(int(rng.integers(1, 60))) + } + keys = _words(*decs) + env = hive.build_root_coverage(keys, 4) + np.testing.assert_array_equal(hive.root_coverage_words(env), np.sort(keys)) + + def test_wrong_order_key_rejected(self): + with pytest.raises(ValueError, match="shard order"): + hive.build_root_coverage(_words("-511", "-5112"), 3) + + def test_malformed_range_rejected(self): + env = hive.build_root_coverage(_words("-511"), 2) + env["ranges"] = [["-511", "512"]] # base-crossing run + with pytest.raises(ValueError, match="malformed coverage range"): + hive.root_coverage_words(env) + + def test_payload_bounded_for_contiguous_fleet(self): + # The transport claim (plan question 3): a spatially coherent 50k-shard + # run serializes to a few-KB envelope, far under Lambda's 256 KB + # async-invoke cap that a raw 50k-key list would break. + keys = _words(*("5" + hive._rank_tail(r, 9) for r in range(50_000))) + env = hive.build_root_coverage(keys, 9) + assert len(env["ranges"]) == 1 + assert len(json.dumps(env).encode()) < 4096 + + def test_docs_reference_example_round_trips(self): + # The frozen-spec reference example in docs/hive_layout.md was + # GENERATED by build_root_coverage (PR #208 round 4); parse it + # straight out of the doc so it can never drift from the code. + import re + + doc = (Path(__file__).parent.parent / "docs" / "hive_layout.md").read_text() + blocks = re.findall(r"```json\n(\{.*?\})\n```", doc, flags=re.DOTALL) + (root_example,) = [ + json.loads(b) for b in blocks if json.loads(b).get("encoding") == "ranges" + ] + words = hive.root_coverage_words(root_example) + expected = _words("-4211321", "-4211322", "-4211323", "-4211324", "5112333") + np.testing.assert_array_equal(words, np.sort(expected)) + # And it is literally what the serializer emits for those shards. + rebuilt = hive.build_root_coverage(expected, 6) + assert rebuilt["ranges"] == root_example["ranges"] + assert rebuilt["order"] == root_example["order"] == 6 + + def test_write_creates_then_unions(self, tmp_path): + root = str(tmp_path / "store") + first = hive.build_root_coverage(_words("-511", "-512"), 2) + hive.write_root_coverage(root, first) + stored = hive.read_root_coverage(root) + np.testing.assert_array_equal( + hive.root_coverage_words(stored), np.sort(_words("-511", "-512")) + ) + # Incremental run: a second write UNIONS with the existing object. + second = hive.build_root_coverage(_words("-513", "521"), 2) + hive.write_root_coverage(root, second) + merged = hive.read_root_coverage(root) + np.testing.assert_array_equal( + hive.root_coverage_words(merged), + np.sort(_words("-511", "-512", "-513", "521")), + ) + # The adjacent -511..-513 accumulate into one run across runs. + assert ["-511", "-513"] in merged["ranges"] + + def test_unparsable_existing_is_overwritten(self, tmp_path): + # The root object is a regenerable cache (D9): garbage is logged and + # replaced, never merged — the sweep is the authoritative rebuilder. + root = tmp_path / "store" + root.mkdir() + (root / hive.ROOT_COVERAGE_NAME).write_bytes(b"not json {") + env = hive.build_root_coverage(_words("-511"), 2) + hive.write_root_coverage(str(root), env) + assert hive.read_root_coverage(str(root))["ranges"] == [["-511", "-511"]] + + def test_incompatible_existing_is_overwritten(self, tmp_path): + root = str(tmp_path / "store") + hive.write_root_coverage(root, hive.build_root_coverage(_words("-5112"), 3)) + env = hive.build_root_coverage(_words("-511"), 2) # different order + hive.write_root_coverage(root, env) + assert hive.read_root_coverage(root)["order"] == 2 + assert hive.read_root_coverage(root)["ranges"] == [["-511", "-511"]] + + +def _local_agg_catalog(tmp_path, shard): + catalog = { + "metadata": {"short_name": "ATL06", "version": "007"}, + "grid_signature": { + "type": "healpix", + "indexing_scheme": "nested", + "parent_order": 6, + "child_order": 12, + "layout": "fullsphere", + }, + "shard_keys": [int(shard)], + "granules": [[{"id": "g1", "s3": "s3://b/g1.h5", "https": "https://h/g1.h5"}]], + } + path = tmp_path / "catalog.json" + path.write_text(json.dumps(catalog)) + return str(path) + + +class TestLocalRootCoverage: + """End-to-end through the local backend: the run loop's successful shard + set becomes the root MOC; failures are excluded; flag-off writes nothing.""" + + def _agg(self, monkeypatch, tmp_path, *, meta_error=None, coverage_moc=None): + from zagg import runner + from zagg.runner import agg + + cfg = default_config("atl06") + cfg.output["store_layout"] = "hive" + if coverage_moc is not None: + cfg.output["coverage_moc"] = coverage_moc + shard = morton_word(SHARD) + monkeypatch.setattr(runner, "get_nsidc_s3_credentials", lambda: {"accessKeyId": "a"}) + + def fake_hive_write(shard_key, granule_urls, grid, s3_creds, store_root, config, **kw): + return {"shard_key": int(shard_key), "error": meta_error, "total_obs": 1} + + monkeypatch.setattr(hive, "process_and_write_hive", fake_hive_write) + root = str(tmp_path / "out") + agg(cfg, catalog=_local_agg_catalog(tmp_path, shard), store=root, backend="local") + return root, shard + + def test_successful_run_writes_root_moc(self, monkeypatch, tmp_path): + root, shard = self._agg(monkeypatch, tmp_path) + env = hive.read_root_coverage(root) + assert env["order"] == 6 + assert env["source"] == "dispatcher" + np.testing.assert_array_equal( + hive.root_coverage_words(env), np.asarray([shard], dtype=np.uint64) + ) + + def test_failed_shards_excluded(self, monkeypatch, tmp_path): + # Every shard errored -> no successful completions -> no root object. + root, _shard = self._agg(monkeypatch, tmp_path, meta_error="boom") + assert hive.read_root_coverage(root) is None + + def test_flag_off_writes_nothing(self, monkeypatch, tmp_path): + import os + + root, _shard = self._agg(monkeypatch, tmp_path, coverage_moc=False) + assert hive.read_root_coverage(root) is None + assert sorted(os.listdir(root)) == [hive.MANIFEST_NAME] + + +class TestLambdaCoverageDispatch: + """The dispatcher leg: ONE fire-and-forget invoke with the serialized + envelope after the fan-out concludes; byte-identical dispatch when off; + fail-open when the invoke raises.""" + + def _agg(self, monkeypatch, tmp_path, captured, *, coverage_moc=None, invoke=None): + from unittest.mock import MagicMock + + import boto3 + + from zagg import runner + from zagg.concurrency import ConcurrencyReport + from zagg.runner import agg + + cfg = default_config("atl06") + cfg.output["store_layout"] = "hive" + if coverage_moc is not None: + cfg.output["coverage_moc"] = coverage_moc + shard = morton_word(SHARD) + catalog_path = _local_agg_catalog(tmp_path, shard) + + monkeypatch.setattr( + runner, + "get_nsidc_s3_credentials", + lambda: {"accessKeyId": "a", "secretAccessKey": "s", "sessionToken": "t"}, + ) + monkeypatch.setattr(boto3, "Session", lambda *a, **k: MagicMock()) + monkeypatch.setattr(runner, "_get_function_timeout_s", lambda *a, **k: 720) + monkeypatch.setattr( + runner, + "compute_available_workers", + lambda requested, *a, **k: ( + 1, + ConcurrencyReport( + account_limit=1000, + current_concurrent=0, + padding=100, + available=900, + function_reserved=None, + ), + ), + ) + monkeypatch.setattr(runner, "_invoke_lambda_setup", lambda *a, **kw: None) + monkeypatch.setattr(runner, "_invoke_lambda_finalize", lambda *a, **k: None) + monkeypatch.setattr( + runner, + "_invoke_lambda_cell", + lambda *a, **k: { + "status_code": 200, + "body": {"total_obs": 1}, + "error": None, + "lambda_duration": 1.0, + "shard_key": shard, + }, + ) + monkeypatch.setattr( + runner, + "_invoke_lambda_coverage", + invoke + or (lambda client, fn, store, envelope, **kw: captured.update(envelope=envelope)), + ) + summary = agg(cfg, catalog=catalog_path, store="s3://out/product", backend="lambda") + return summary, shard + + def test_dispatches_serialized_envelope(self, monkeypatch, tmp_path): + captured: dict = {} + _summary, shard = self._agg(monkeypatch, tmp_path, captured) + env = captured["envelope"] + assert env["encoding"] == "ranges" and env["order"] == 6 + np.testing.assert_array_equal( + hive.root_coverage_words(env), np.asarray([shard], dtype=np.uint64) + ) + + def test_flag_off_dispatch_is_byte_identical(self, monkeypatch, tmp_path): + captured: dict = {} + self._agg(monkeypatch, tmp_path, captured, coverage_moc=False) + assert captured == {} # no coverage invoke at all + + def test_invoke_failure_is_fail_open(self, monkeypatch, tmp_path): + def boom(*a, **k): + raise RuntimeError("event invoke failed") + + summary, _shard = self._agg(monkeypatch, tmp_path, {}, invoke=boom) + assert summary["cells_with_data"] == 1 # the run result is untouched + + +class TestHandlerCoverageMode: + """The worker leg: ``mode: "coverage"`` GET-unions-PUTs the root object.""" + + @staticmethod + def _ctx(): + ctx = MagicMock() + ctx.aws_request_id = "req-1" + ctx.function_name = "process-shard" + ctx.memory_limit_in_mb = 2048 + ctx.get_remaining_time_in_millis.return_value = 900_000 + return ctx + + def _event(self, root, envelope): + return {"mode": "coverage", "store_path": root, "coverage": envelope} + + def test_writes_root_object(self, handler_mod, tmp_path): + root = str(tmp_path / "store") + env = hive.build_root_coverage(_words(SHARD), 6) + resp = handler_mod.lambda_handler(self._event(root, env), self._ctx()) + assert resp["statusCode"] == 200 + assert json.loads(resp["body"])["mode"] == "coverage" + stored = hive.read_root_coverage(root) + np.testing.assert_array_equal( + hive.root_coverage_words(stored), hive.root_coverage_words(env) + ) + + def test_unions_with_existing(self, handler_mod, tmp_path): + root = str(tmp_path / "store") + prior = SHARD[:-1] + "4" # sibling shard from an earlier run + hive.write_root_coverage(root, hive.build_root_coverage(_words(prior), 6)) + env = hive.build_root_coverage(_words(SHARD), 6) + resp = handler_mod.lambda_handler(self._event(root, env), self._ctx()) + assert resp["statusCode"] == 200 + np.testing.assert_array_equal( + hive.root_coverage_words(hive.read_root_coverage(root)), + np.sort(_words(SHARD, prior)), + ) + + def test_error_returns_500_never_raises(self, handler_mod, tmp_path): + # Fail-open contract: nobody reads the Event-invoke response, but the + # handler must degrade to a logged 500, not an unhandled exception. + event = self._event(str(tmp_path / "store"), hive.build_root_coverage(_words(SHARD), 6)) + del event["coverage"] # malformed event + resp = handler_mod.lambda_handler(event, self._ctx()) + assert resp["statusCode"] == 500 + assert json.loads(resp["body"])["mode"] == "coverage" + # And a malformed EXISTING object never breaks the write: it is + # overwritten (regenerable cache), still a 200. + root = tmp_path / "store2" + root.mkdir() + (root / hive.ROOT_COVERAGE_NAME).write_bytes(b"garbage {") + ok = handler_mod.lambda_handler( + self._event(str(root), hive.build_root_coverage(_words(SHARD), 6)), self._ctx() + ) + assert ok["statusCode"] == 200 + assert hive.read_root_coverage(str(root))["ranges"] == [[SHARD, SHARD]] + + +# ── reader primitives (phase 4) ────────────────────────────────────────────── + + +@pytest.fixture(autouse=True) +def _fresh_staleness_dedup(): + """The warn-once dedup is process-global by design; isolate tests.""" + from zagg import coverage + + coverage._stale_warned.clear() + yield + coverage._stale_warned.clear() + + +class TestLoadCoverage: + def test_present_envelope_loads(self, tmp_path): + from zagg.coverage import load_coverage + + root = str(tmp_path / "store") + hive.write_root_coverage(root, hive.build_root_coverage(_words(SHARD), 6)) + env = load_coverage(root) + assert env["encoding"] == "ranges" and env["ranges"] == [[SHARD, SHARD]] + + def test_missing_reads_none(self, tmp_path): + from zagg.coverage import load_coverage + + assert load_coverage(str(tmp_path / "empty")) is None + + def test_garbage_reads_none_never_raises(self, tmp_path): + from zagg.coverage import load_coverage + + root = tmp_path / "store" + root.mkdir() + (root / hive.ROOT_COVERAGE_NAME).write_bytes(b"garbage {") + assert load_coverage(str(root)) is None + + def test_unknown_spec_or_encoding_reads_none(self, tmp_path): + from zagg.coverage import load_coverage + + root = tmp_path / "store" + root.mkdir() + env = hive.build_root_coverage(_words(SHARD), 6) + for mutation in ({"spec": "morton-moc/2"}, {"encoding": "bitmap"}): + (root / hive.ROOT_COVERAGE_NAME).write_text(json.dumps({**env, **mutation})) + assert load_coverage(str(root)) is None + + +class TestIntersections: + def _leaf_cov(self, *decimals): + return hive.build_coverage(morton_word(SHARD), _words(*decimals), 8) + + def test_root_coverage_and_hit_and_miss(self): + from zagg.coverage import root_coverage_and + + env = hive.build_root_coverage(_words(SHARD, "-5112334"), 6) + # An AOI at CELL order inside a covered shard intersects (mixed-order + # containment via moc_and)... + hit = root_coverage_and(env, _words(SHARD + "12")) + assert hit.size > 0 + # ...a disjoint sibling misses. + assert root_coverage_and(env, _words("-5112331")).size == 0 + + def test_root_coverage_and_boundary(self): + from zagg.coverage import root_coverage_and + + env = hive.build_root_coverage(_words(SHARD), 6) + # The covered shard itself (range endpoint) intersects exactly. + np.testing.assert_array_equal(root_coverage_and(env, _words(SHARD)), _words(SHARD)) + + def test_box_and_hit_and_miss(self): + from zagg.coverage import box_and + + cov = self._leaf_cov(SHARD + "12", SHARD + "43") + assert box_and(cov, _words(SHARD + "12")).size > 0 + assert box_and(cov, _words(SHARD + "121")).size > 0 # deeper AOI cell + assert box_and(cov, _words(SHARD + "2")).size == 0 # outside both members + + def test_box_and_skips_null_slots(self): + from zagg.coverage import box_and + + cov = self._leaf_cov(SHARD + "12") # 1 member + 3 nulls + assert box_and(cov, _words(SHARD + "12")).size > 0 + + def test_bitmap_and_exact_and_none_fallback(self, monkeypatch, tmp_path): + import zagg.processing as processing + from zagg.coverage import bitmap_and + from zagg.grids import HealpixGrid + + cfg = default_config("atl06") + grid = HealpixGrid(parent_order=6, child_order=8, layout="fullsphere", config=cfg) + word = morton_word(SHARD) + occupied = _words(SHARD + "12", SHARD + "43") + + def fake(g, shard_key, urls, **kwargs): + import pandas as pd + + from zagg.config import get_data_vars + + coords = grid.chunk_coords(shard_key) + df = pd.DataFrame( + {var: 0.0 for var in get_data_vars(cfg)}, index=range(len(coords["cell_ids"])) + ) + for name, vals in coords.items(): + df[name] = vals + kwargs["write_chunk"](grid.block_index(int(shard_key)), df, {}) + kwargs["occupied_out"].append(occupied) + return pd.DataFrame(), { + "shard_key": int(shard_key), + "cells_with_data": 2, + "total_obs": 2, + "granule_count": 1, + "files_processed": 1, + "duration_s": 0.0, + "error": None, + } + + monkeypatch.setattr(processing, "process_shard", fake) + root = str(tmp_path / "store") + hive.process_and_write_hive(word, ["s3://b/g.h5"], grid, {}, root, cfg, store_kwargs={}) + leaf = hive.shard_leaf_path(root, word) + # Exact: an occupied cell intersects; an unoccupied one is a + # DEFINITIVE miss (the bitmap is exact, unlike the box). + assert bitmap_and(leaf, _words(SHARD + "12")).size > 0 + assert bitmap_and(leaf, _words(SHARD + "21")).size == 0 + # Box-only leaf (no sidecar): None -> caller falls back to the box. + empty_leaf = str(tmp_path / "nothing.zarr") + assert bitmap_and(empty_leaf, _words(SHARD + "12")) is None + + +class TestWarnIfStale: + def test_warns_once_per_store_on_mismatch(self, tmp_path): + from zagg.coverage import warn_if_stale + + env = hive.build_root_coverage(_words("-5112334"), 6) # sibling only + root = str(tmp_path / "store") + with pytest.warns(UserWarning, match="refresh_root_coverage"): + assert warn_if_stale(root, morton_word(SHARD), env) is True + # Second mismatch on the SAME store: still stale, but silent. + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert warn_if_stale(root, morton_word(SHARD), env) is True + # A DIFFERENT store warns independently. + with pytest.warns(UserWarning): + assert warn_if_stale(str(tmp_path / "other"), morton_word(SHARD), env) is True + + def test_latch_key_is_slash_normalized(self, tmp_path): + # root and root/ are the same store: one warning, not two (round 4). + from zagg.coverage import warn_if_stale + + env = hive.build_root_coverage(_words("-5112334"), 6) + root = str(tmp_path / "store") + with pytest.warns(UserWarning): + warn_if_stale(root, morton_word(SHARD), env) + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert warn_if_stale(root + "/", morton_word(SHARD), env) is True + + def test_refresh_rearms_the_latch(self, tmp_path): + # Once per stale EPISODE (round 4): after a successful refresh, a new + # mismatch on the same store warns again. + from zagg import coverage + from zagg.coverage import warn_if_stale + from zagg.grids import HealpixGrid + from zagg.store import open_store + + cfg = default_config("atl06") + grid = HealpixGrid(parent_order=6, child_order=8, layout="fullsphere", config=cfg) + root = str(tmp_path / "store") + hive.ensure_manifest(root, hive.build_manifest(grid)) + leaf = hive.shard_leaf_path(root, morton_word(SHARD)) + store = open_store(leaf) + grid.emit_shard_template(store, overwrite=True) + hive.stamp_commit(store, cells_with_data=1, granule_count=1) + + stale_env = hive.build_root_coverage(_words("-5112334"), 6) + with pytest.warns(UserWarning): + warn_if_stale(root, morton_word(SHARD), stale_env) + coverage.refresh_root_coverage(root) + # New episode on the SAME store: the latch was re-armed. + with pytest.warns(UserWarning): + assert warn_if_stale(root, morton_word(SHARD), stale_env) is True + + def test_listed_shard_is_not_stale(self, tmp_path): + from zagg.coverage import warn_if_stale + + env = hive.build_root_coverage(_words(SHARD, "-5112334"), 6) + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert warn_if_stale(str(tmp_path), morton_word(SHARD), env) is False + + def test_no_root_moc_is_absence_not_staleness(self, tmp_path): + from zagg.coverage import warn_if_stale + + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert warn_if_stale(str(tmp_path), morton_word(SHARD), None) is False + + def test_malformed_envelope_counts_as_stale(self, tmp_path): + from zagg.coverage import warn_if_stale + + with pytest.warns(UserWarning): + assert warn_if_stale(str(tmp_path), morton_word(SHARD), {"order": 6}) is True + + +class TestRefreshRootCoverage: + def _store_with_leaves(self, tmp_path, stamped, debris=()): + from zagg.grids import HealpixGrid + from zagg.store import open_store + + cfg = default_config("atl06") + grid = HealpixGrid(parent_order=6, child_order=8, layout="fullsphere", config=cfg) + root = str(tmp_path / "store") + hive.ensure_manifest(root, hive.build_manifest(grid)) + for dec in (*stamped, *debris): + leaf = hive.shard_leaf_path(root, morton_word(dec)) + store = open_store(leaf) + grid.emit_shard_template(store, overwrite=True) + if dec in stamped: + hive.stamp_commit(store, cells_with_data=1, granule_count=1) + return root + + def test_rebuilds_exactly_the_stamped_set(self, tmp_path): + from zagg.coverage import refresh_root_coverage + + # Both hemispheres (two base dirs) + one unstamped debris leaf. + root = self._store_with_leaves( + tmp_path, stamped=(SHARD, "-5112334", "5112333"), debris=("-5112331",) + ) + env = refresh_root_coverage(root) + assert env["source"] == "refresh" + assert env["order"] == 6 + np.testing.assert_array_equal( + hive.root_coverage_words(env), + np.sort(_words(SHARD, "-5112334", "5112333")), + ) + # The written object matches the returned envelope. + assert hive.read_root_coverage(root) == env + + def test_replaces_stale_root_object(self, tmp_path): + from zagg.coverage import refresh_root_coverage + + root = self._store_with_leaves(tmp_path, stamped=(SHARD,)) + # A stale root claims a shard that has no stamped leaf. + hive.write_root_coverage(root, hive.build_root_coverage(_words("-5112334"), 6)) + env = refresh_root_coverage(root) + # No union: the walk is ground truth and SUPERSEDES the stale cache. + np.testing.assert_array_equal(hive.root_coverage_words(env), _words(SHARD)) + + def test_no_stamped_leaves_removes_the_cache(self, tmp_path): + from zagg.coverage import refresh_root_coverage + + root = self._store_with_leaves(tmp_path, stamped=(), debris=(SHARD,)) + hive.write_root_coverage(root, hive.build_root_coverage(_words(SHARD), 6)) + assert refresh_root_coverage(root) is None + assert hive.read_root_coverage(root) is None + + def test_foreign_order_leaf_skipped_with_warning(self, tmp_path, caplog): + # A stamped leaf at a non-manifest order (old-config survivor or a + # hand-copied leaf) must not kill the escape hatch (round 4): it is + # skipped with a logged warning and the root MOC is rebuilt from the + # conforming set. + import logging + + from zagg.coverage import refresh_root_coverage + from zagg.grids import HealpixGrid + from zagg.store import open_store + + root = self._store_with_leaves(tmp_path, stamped=(SHARD,)) + cfg = default_config("atl06") + alien_grid = HealpixGrid(parent_order=5, child_order=8, layout="fullsphere", config=cfg) + alien = SHARD[:-1] # order-5 ancestor of the legit shard's sibling space + leaf = hive.shard_leaf_path(root, morton_word(alien)) + store = open_store(leaf) + alien_grid.emit_shard_template(store, overwrite=True) + hive.stamp_commit(store, cells_with_data=1, granule_count=1) + + with caplog.at_level(logging.WARNING, logger="zagg.coverage"): + env = refresh_root_coverage(root) + assert any("mixed-order stores are unsupported" in r.message for r in caplog.records) + np.testing.assert_array_equal(hive.root_coverage_words(env), _words(SHARD)) + + def test_not_a_hive_root_raises(self, tmp_path): + from zagg.coverage import refresh_root_coverage + + with pytest.raises(ValueError, match="not a hive store root"): + refresh_root_coverage(str(tmp_path / "nowhere")) diff --git a/tests/test_hive.py b/tests/test_hive.py index d187638..ad793ee 100644 --- a/tests/test_hive.py +++ b/tests/test_hive.py @@ -331,10 +331,12 @@ def _run(self, monkeypatch, cfg, tmp_path, fake): ) return grid, shard, root, meta - def _streaming_fake(self, grid, ragged=None): + def _streaming_fake(self, grid, ragged=None, occupied=None): def fake(g, shard_key, urls, **kwargs): carrier = self._carrier(grid, shard_key) kwargs["write_chunk"](grid.block_index(int(shard_key)), carrier, ragged or {}) + if occupied is not None and kwargs.get("occupied_out") is not None: + kwargs["occupied_out"].append(np.asarray(occupied, dtype=np.uint64)) return pd.DataFrame(), self._meta(shard_key) return fake @@ -405,8 +407,17 @@ def torn(g, shard_key, urls, **kwargs): ) assert os.path.exists(leaf) # the prefix exists... assert hive.read_commit(open_store(leaf)) is None # ...but is debris + # No stamp -> no coverage visible either (issue #200): the tier-0 + # payload rides the stamp, so a torn worker never publishes coverage. + assert hive.read_coverage(open_store(leaf)) is None stale = os.path.join(leaf, grid.group_path, "h", morton_decimal(shard)) assert os.path.exists(stale) # the torn attempt's CSR subgroup + # Plant a sidecar in the debris: the one leaf object zarr does NOT + # own must also fall to the wholesale wipe (PR #208 round 2) — this + # goes red if the re-template ever drifts to node-by-node rewrites. + hive.write_coverage_sidecar(leaf, b"torn-attempt sidecar") + sidecar = os.path.join(leaf, hive.COVERAGE_SIDECAR) + assert os.path.exists(sidecar) # Retry (no ragged this time): same leaf, overwritten wholesale — # the stale subgroup is GONE — and stamped at the end. @@ -416,6 +427,7 @@ def torn(g, shard_key, urls, **kwargs): ) assert hive.read_commit(open_store(leaf))["complete"] is True assert not os.path.exists(stale), "stale torn-write object survived the re-template" + assert not os.path.exists(sidecar), "torn attempt's sidecar survived the re-template" def test_errored_shard_is_not_stamped(self, monkeypatch, cfg, tmp_path): # A shard that wrote chunks but ended in error stays unstamped debris. @@ -477,7 +489,12 @@ def wrapped(*a, **k): return wrapped grid = self._grid(cfg) - fake = self._streaming_fake(grid, ragged={"h": ([np.array([1.0])], [0])}) + shard = _shard_word() + fake = self._streaming_fake( + grid, + ragged={"h": ([np.array([1.0])], [0])}, + occupied=grid.children(shard)[:2], + ) monkeypatch.setattr(processing, "process_shard", fake) monkeypatch.setattr( processing, "write_dataframe_to_zarr", rec("dense", processing.write_dataframe_to_zarr) @@ -485,11 +502,17 @@ def wrapped(*a, **k): monkeypatch.setattr( processing, "write_ragged_to_zarr", rec("ragged", processing.write_ragged_to_zarr) ) + monkeypatch.setattr( + hive, "write_coverage_sidecar", rec("sidecar", hive.write_coverage_sidecar) + ) monkeypatch.setattr(hive, "stamp_commit", rec("stamp", hive.stamp_commit)) hive.process_and_write_hive( - _shard_word(), ["s3://b/g1.h5"], grid, {}, str(tmp_path / "store"), cfg, store_kwargs={} + shard, ["s3://b/g1.h5"], grid, {}, str(tmp_path / "store"), cfg, store_kwargs={} ) - assert ops == ["dense", "ragged", "stamp"] + # The coverage sidecar (issue #200 phase 2) lands BEFORE the stamp: + # the stamp stays the leaf's final write, so an unstamped prefix's + # sidecar is debris like everything else in it. + assert ops == ["dense", "ragged", "sidecar", "stamp"] class TestLeafBlockIndex: @@ -549,8 +572,10 @@ def fake_hive_write(shard_key, granule_urls, grid, s3_creds, store_root, config, agg(cfg, catalog=catalog_path, store=root, backend="local") assert calls == [(shard, root)] - # Template time wrote ONLY the manifest — no shared zarr template (D5). - assert sorted(os.listdir(root)) == [hive.MANIFEST_NAME] + # Template time wrote ONLY the manifest — no shared zarr template + # (D5). The end-of-run root coverage.moc (issue #200 phase 3, + # default-on for hive) is the only other root object. + assert sorted(os.listdir(root)) == [hive.ROOT_COVERAGE_NAME, hive.MANIFEST_NAME] assert hive.read_manifest(root)["shard_order"] == 6 def test_lambda_hive_dispatches_with_manifest_dataset(self, monkeypatch, cfg, tmp_path):