Inline selection-read heuristic + granule_workers default 4

[espg]

Closes #185. Refs #184 (the per-cell clamp below is item 1's simple min() case — the rest of #184 stays open). Refs #186 (ask (1) only — root-cause of the sharded write race stays on #186).

Implements option (2) from the issue thread — the inline-internal heuristic — plus the default granule_workers bump, per the approval on issue #185 ("yup, sounds like (2) it is (plus the default granule workers bump)"). The mechanism and evidence are the K-sweep tables on PR #183: #179's selection routing costs inline ~41% at K=1 (B-tree chunk-map walks for the segment-rate selection datasets) while being a pure win for sidecar.

Phases

• Phase 1 — inline selection-read heuristic (src/zagg/index/inline.py)
• Phase 2 — granule_workers default 1 → 4 (src/zagg/processing/worker.py, README)
• Phase 3 — dispatcher-side per-cell min(K, n_granules) clamp (src/zagg/runner.py)
• Phase 4 — surface the chained CSR write failure (issue Sharded-path parallel CSR writes race under real CPU parallelism (fails 3/3 on 6 GB workers, largest shard) #186 ask (1), src/zagg/processing/write.py)
• Fold adversarial-review findings (review): clamp on resolved-URL count; bounded per-failure write summary
• Fold second-round review findings (review): header-first failure roll call + single-failure guard; named _RAGGED_FAILURE_LOG_CAP; fd_safe_max_workers pins in the fanout failure tests; lockstep comment on the local clamp resolve

What / approach

Phase 1 — the heuristic, contained in InlineIndex._chunk_aligned_read_fn. read.py's call sites are untouched; the seam stays uniform. The discriminator is the call shape: on the planned route, data reads arrive as hyperslices (execute_read_plan) and selection reads (coarse coords + link arrays, cross-level filter flags, segment-level broadcasts) arrive as full-dataset reads (hyperslice is None). The change is one early exit:

if selection_direct and hyperslice is None:   # selection_direct = planned and not write_back
    return h5obj.readDatasets([path])[path]

• Default path (planned=True, write_back off — the shipped index: None default where the regression lives): unmapped full-dataset reads are served whole through h5coro's readDatasets (its 4 MiB cache-line path, the fast side of the Compiled (hidefix) read path is never exercised by default, and its fetch/decode is serial #170 matrix for these datasets) and build no chunk map — restoring the 0.17 selection behavior.
• write_back ON: unchanged byte-for-byte. The walk's product is the manifest payload (_pending_for/finish_granule), so selection datasets stay mapped and served compiled — freshly built manifests keep covering the selection datasets, preserving sidecar's −25% K=1 win.
• planned=False (flat sources): unchanged in all modes — the compiled full-dataset decode there is the point of issue Compiled (hidefix) read path is never exercised by default, and its fetch/decode is serial #170.
• A path already mapped this read_group (e.g. both a data and a selection read) keeps the compiled route — the map is already paid for. Full readDatasets reads don't need the PR 88S stress shard: pins, 900s timeout, chunk-boundary extraction, a-priori read plan, streaming tdigest (issue #148) #152 start-edge workaround (that's a ranged-read quirk), so no boundary handling is added.

Call-shape note beyond the issue-thread sketch: the plan.full_read selectivity fallback (_planned_read_group → _read_group_full with the planned read_fn, routed through the seam since #179) also issues full-dataset reads — the base-rate coordinate reads. The heuristic serves those direct too, which matches the pre-#179 fallback (it bypassed the seam entirely, per inline.py's own docstring); the fallback's ranged data reads stay compiled. Pinned by test_full_read_fallback_data_reads_stay_compiled.

A/B watch-list (review finding 2, no code change): direct readDatasets reads retain their 4 MiB cache lines until granule close. On the full_read fallback that can be ~245 MB/beam of base-rate volume, × K=4 granules in flight at the new default — no fleet datapoint yet; watch as an issue #117 memory axis in the next sweep.

Phase 2 — default granule_workers: 4. Directed on the PR #183 K-sweep evidence (K=4 inline 296.9 s vs 0.17's 323 s median, o9 NEON AOI) via the same approval comment. Docs updated honestly: output remains byte-identical to serial by construction of the ordered fold — the default just no longer equals serial execution (granule_workers: 1 restores the serial loop, which still constructs no pool).

Phase 3 — dispatcher-side clamp (Refs #184, item 1's simple case). _clamped_data_source(data_source, n_granules) in runner.py returns a shallow-copied data_source carrying min(K, n_granules) only when that lowers K, else None. Both _cell_work sites use it: _run_local via dataclasses.replace(config, data_source=...), _run_lambda via a per-cell {**config_dict, "data_source": ...}. Unclamped cells pass the shared config through untouched, keeping their event payloads byte-identical. Per the review fold, the clamp counts the resolved URLs (_resolve_urls output — what the worker actually reads), not raw records, so a partially-resolvable cell no longer under-clamps. Deviation from the plan sketch: there is no dedicated worker event key for this — the worker reads the width from the event's config (_granule_workers(event["config"]["data_source"]), PR #183), so the clamp rides the per-cell config copy rather than a new top-level key (no lambda_handler change needed).

Phase 4 — CSR write-failure observability (issue #186 ask (1) only). _write_ragged_fanout's failure path now (a) logs the first failing subgroup's chained exception with its full traceback (logger.error(..., exc_info=first_exc)), (b) when more than one subgroup failed, follows that header with a bounded one-line-per-failure roll call (subgroup_key: ExcType: message, capped at _RAGGED_FAILURE_LOG_CAP = 20 lines) so the race signature across subgroups is visible, and (c) carries TypeName: message inside the raised summary string, so the worker envelope (lambda_handler's Failed to write zarr: {e}) shows the actual S3/zarr error even without log access. Envelope schema unchanged; no behavior change on the write path itself — the root-cause/race fix stays on #186.

How tested

• Phase 1 (tests/test_index.py): test_default_selection_reads_skip_chunk_maps spies the map-build seam (build_chunk_map) and a readDatasets recorder — selection paths are never walked, are served as whole-dataset readDatasets calls, data paths still map + compile, rows identical to the hierarchical baseline. test_write_back_selection_reads_stay_compiled pins the ON mode (readDatasets never touched, pending maps cover the selection datasets, rows identical). test_manifest_byte_identical_to_direct_chunk_maps pins the sidecar-build product: the persisted parquet is byte-identical to one assembled directly from build_chunk_map over the deterministic coverage. Segment-variable, cross-level-filter, full-read-fallback, and per-dataset-degrade tests updated to the new mode split.
• Phase 2 (tests/test_processing.py): default asserted as 4; test_k1_never_touches_the_pool keeps the serial guarantee on the explicit escape hatch; test_default_output_identical_to_serial pins default-vs-serial frame equality under out-of-order completions.
• Phase 3 (tests/test_runner.py): TestClampedDataSource unit-tests the min() policy (incl. default-K and the 0-granule floor); dispatcher tests for both backends assert per-shard event/config values (2-granule cell → 2, 1-granule cell → 1, 6-granule cell → untouched, and a 6-record cell with only 3 s3-resolvable hrefs → 3) and that unclamped local cells receive the shared config object as-is.
• Phase 4 (tests/test_processing.py): the fanout failure test asserts the raised summary matches first failing shard_key 3: OSError: disk full, that the log record carries exc_info whose formatted traceback names the raising frame, and that a single failure logs no roll call (the header alone); test_fanout_logs_each_failure_bounded pins the multi-failure roll call (header first, then 25 failures → _RAGGED_FAILURE_LOG_CAP per-key lines + "and 5 more", exactly one traceback record). Both tests pin fd_safe_max_workers so a low-ulimit host can't silently take the serial branch.
• Full local gate: pytest -q green — 1516 passed, 26 skipped (3.12, uv sync --extra test) on the first-fold state; the second fold re-ran tests/test_index.py tests/test_processing.py tests/test_runner.py (425 passed) plus the two changed modules in full (352 passed); ruff check + ruff format --check clean on every touched file.

Questions for review

1. Concurrent readDatasets on one h5obj: selection reads now reach readDatasets through the read_workers pool (_read_paths_pooled), so up to four single-path calls can be in flight on one granule handle. Answered safe by the adversarial review's h5coro source inspection: h5promise.py already fans one thread per dataset over the shared resource, cache-line access is locked per line, and the S3 client is documented thread-safe.
2. Pre-existing lint drift (flagged, not fixed, per the §4 convention): on clean main, ruff check src tests fails N818 (UnknownCapability, src/zagg/registry.py:64) and ruff format --check would reformat 7 untouched files (src/zagg/__init__.py, src/zagg/__main__.py, src/zagg/catalog/sources.py, src/zagg/viz/crs.py, tests/test_catalog.py, tests/test_extract.py, tests/test_integration.py). The PR lint bot selects only E,F,W,I, so PRs won't see these.
3. The K=4 default reaches direct process_shard callers (no dispatcher) unclamped — a 1-granule hand-rolled call now constructs a 4-wide pool (threads spawn lazily, so the cost is one idle pool object). Worth a worker-side min(K, len(granule_urls)) too, or keep the clamp dispatcher-only as agreed on Follow-ons to #183: dispatcher-side shard-aware granule_workers, and function-size variants gated on utilization #184?

[espg]

🤖 from Claude (review)

Verdict: sound. No blocking-correctness findings across all four phases. The scary claims were verified empirically, not just by reading: I checked out the branch in a scratch clone, ran the touched test files in full — tests/test_index.py + tests/test_processing.py + tests/test_runner.py, 424 passed, 0 failed (which also means the pre-existing default-config process_shard tests survived silently running K=4 pooled) — and byte-compared a write_back manifest produced by base-commit inline.py (dfb7593) against the branch: byte-identical, so the sidecar-build product (the reason option 2 beat option 1 on issue #185) is preserved beyond what the in-tree test asserts.

Verified by inspection: (a) write_back ON is untouched (selection_direct = planned and not self.write_back gates the early exit off; maps still aliases _pending_for); (b) planned=False unreachable by the heuristic; (c) both orders of a path that is both data- and selection-read resolve correctly (maps.get runs first, the direct branch fires only for unmapped no-hyperslice reads); (d) the discriminator holds — under inline, _execute_plan_group only runs with full_read=False, so every hyperslice=None call on the planned route really is a selection/coordinate full read, and full readDatasets reads don't need the PR #152 start-edge workaround (a ranged-read quirk); (e) dtype/row parity is pinned against the hierarchical baseline by assert_frame_equal in every mode split; (f) phase 2's default lives in exactly two places (_granule_workers, README) and both moved — validate_config never encoded the default; (g) phase 3's per-cell copy only diverges on granule_workers (unclamped cells ship the shared config_dict by reference), payload growth is one small dict per clamped cell; (h) phase 4 changes no behavior on the write path, envelope schema unchanged, and the test asserts the traceback frame. Commits are title-only per convention; no new deps; no lint suppressions.

Findings, by severity:

Should-fix

1. src/zagg/runner.py:982 (both _cell_work sites) — the clamp counts len(records) but the worker receives _resolve_urls(records, driver), which drops href-less records; partially-resolvable cells under-clamp. Cheap fix, or fold into your own Question 3 (worker-side clamp covers it and direct callers at once).
2. src/zagg/index/inline.py:552 — direct reads retain h5obj cache lines until granule close where the compiled route freed buffers after decode; on the full_read fallback that's base-rate coordinate volumes, multiplied by the new K=4 in-flight default. No fleet datapoint covers this combination yet — add to the A/B RSS watch list (issue #117 axis).

Answered (your Question 1)
3. src/zagg/index/inline.py:608 — concurrent readDatasets on one handle is structurally identical to the concurrency a single batched call already creates internally (H5Promise fans one thread per dataset; per-cache-line locks; thread-safe S3 client per h5coro's own contract). Safe as-is; details inline.

Nit
4. src/zagg/processing/write.py:443 — consider a one-line-per-failure type/message summary (traceback still first-only), and "first failing" is as_completed-ordered, i.e. nondeterministic.

On your Question 3: a worker-side min(K, len(granule_urls)) would subsume finding (1) and costs one line inside process_shard's already-resolved guard — but it changes the worker/dispatcher contract split agreed on issue #184 item 1, so that call stays with the repo owner.

---

Generated by Claude Code

[espg]

🤖 from Claude (review)

Nit (take or leave): only the first failure's cause is surfaced, and "first" is as_completed order. Issue #186's three reproductions each failed 1-2 of 256 subgroups, and for a race-signature bug the distribution of causes is diagnostic (all the same exception type/message, or mixed?). Since errors already holds every (key, exc), a one-line-per-failure type(exc).__name__: exc summary (full traceback still only for the first, to stay clear of the issue #175 metric-filter concern the neighboring finish_granule handler documents) would cost ~3 lines and make the next 6 GB reproduction answer the "same race everywhere?" question in one shot. Also note errors[0] is completion-ordered, so "first failing shard_key" is nondeterministic across runs — pre-existing wording, but now that the summary string is the operator's primary signal it may be worth saying "e.g." rather than "first". Both fine to decline; the ask (1) contract — type + message + traceback visible in log and envelope — is met, and the test pins the traceback frame (in boom), not just the repr.

---

Generated by Claude Code

[espg]

🤖 from Claude

Partially folded in c65a7c8: the per-failure roll call landed (subgroup_key: ExcType: message, capped at 20 lines + "and N more", one traceback for the first failure; pinned by test_fanout_logs_each_failure_bounded). The "first → e.g." wording change is declined as offered — the summary string keeps "first failing" and the completion-order caveat stands. Leaving this thread open for that declined wording point.

---

Generated by Claude Code

[espg]

🤖 from Claude

Adversarial-review findings folded in c65a7c8 (fold review: clamp on resolved urls + per-failure write summary):

• Clamp counts resolved URLs (finding): both _cell_work sites now clamp on len(_resolve_urls(records, ...)) — the granules the worker actually reads — with the lambda site hoisting the already-computed list. The dispatcher test's catalog gained a 6-record cell with only 3 s3-resolvable hrefs, asserted to clamp to 3.
• Per-failure write summary (finding): _write_ragged_fanout now logs a bounded one-line-per-failure roll call (subgroup_key: ExcType: message, capped at 20 lines + a "N more" tail), keeping the single full traceback for the first-completed failure (the issue main red after #170 merge + template hammer edits (test_apriori/test_streaming under the inline default; template pins) #175 metric-filter concern). New test pins 25 failures → 20 lines + tail + exactly one exc_info record. Declined only the "first → e.g." wording tweak; the roll call now shows every key, so the summary string's example key is no longer the only signal.
• Question 1 (pooled readDatasets) marked answered-safe in the body, citing the review's h5coro source inspection; the RSS watch-list note (direct reads retain 4 MiB cache lines until granule close; full_read fallback × K=4 has no fleet datapoint, issue Order-10 single-shard processing OOMs at 2 GB (gain_bias) — 2 GB is marginal for order-10 #117 axis) added to the phase 1 section — no code change.

Local gate after the fold: pytest -q 1516 passed / 26 skipped; ruff check + format clean on touched files.

---

Generated by Claude Code

[espg]

🤖 from Claude (review)

Fresh-context adversarial review of fold commit c65a7c8 ("fold review: clamp on resolved urls + per-failure write summary") — the delta since the earlier review of 6ad62ea..448e815.

Verdict: the fold is correct; no must-fix findings. Both prior findings are properly addressed:

• Resolved-URL clamp: both dispatch paths now clamp on len(_resolve_urls(...)) — what the worker actually reads. The lambda path dedupes the resolution into granule_urls reused for the invoke, guaranteeing clamp and payload can't diverge. Edge cases check out: zero resolvable URLs hits the pre-existing max(n, 1) guard (unit-pinned at tests/test_runner.py line 1141), K-already-lower and at/above-K pass-through are unchanged, and non-mutation of the shared data_source is pinned (line 1132). The shared-config byte-identity for unclamped cells is preserved (ds is None → shared object/dict as-is, asserted in both integration tests).
• Bounded roll call: exactly min(N, 20) one-line summaries + "and N−20 more", traceback (exc_info) on the single header record only, raised summary string unchanged, no behavior change to the write path itself. The new test pins the cap (25 failures → 20 lines + "and 5 more"), per-key error attribution, and the single-traceback invariant.

Three nits inline (all diff-scoped, none blocking): duplicate URL resolution on the local path vs the deduped lambda path; roll-call log lines emitted before the header they visually continue (plus single-failure duplication); bare 20 literal vs a named constant; and the new fanout test not pinning fd_safe_max_workers the way its neighbor deliberately does.

Checked/ran: git show c65a7c8 against the full PR diff; uv run pytest tests/test_runner.py tests/test_processing.py — 352 passed; ruff check + ruff format --check on the four touched files — clean. Touched src modules within limits (write.py 789 lines; runner.py's pre-existing size was covered in the earlier review and the fold adds ~11 net lines).

---

Generated by Claude Code

[espg]

🤖 from Claude

Status: all phases complete, both review rounds folded, CI green on c2fecab — awaiting espg review. This run:

• Ran the fresh-context adversarial review of the first fold commit c65a7c8 (review): verdict sound, no must-fix findings, four nits.
• Folded all four in c2fecab: header-first failure logging with a len(errors) > 1 guard (single failure logs exactly the pre-fold output), the cap named as _RAGGED_FAILURE_LOG_CAP, fd_safe_max_workers pins in both fanout failure tests, and a lockstep comment on the local clamp's duplicate _resolve_urls (the signature ripple wasn't worth it — _process_and_write has direct test callers). Replies on each inline thread record the dispositions; the one open thread is the declined "first → e.g." wording nit, left standing per convention.
• Verified locally: ruff check + ruff format --check clean; tests/test_processing.py + tests/test_runner.py full modules 352 passed on c2fecab (earlier full pytest -q on the first-fold state: 1516 passed, 26 skipped).

Process note — draft state: repo conventions (CLAUDE.md §2) say a PR in this state — all phases done, findings folded, CI green, not waiting on a question — should be flipped from draft to ready-for-review by the routine itself. This run's routine configuration explicitly forbids marking PRs ready-for-review, and CLAUDE.md's own preamble says to surface such a conflict rather than guess. So the PR stays draft with waiting applied; it is review-ready in substance, and the draft→ready flip (one click, reversible) is left to espg. The open items for review remain Questions 2 (pre-existing lint drift, flagged only) and 3 (worker-side clamp for direct process_shard callers) in the body.

---

Generated by Claude Code