pyramid v2: staged dense sweep on the partition machinery (issue #384) - #416
Conversation
| run_id=run_id, | ||
| run_started=run_started, | ||
| store_kwargs=store_kwargs, | ||
| on_stage=lambda row: heartbeat_lease(store_root, lease, store_kwargs=store_kwargs), |
There was a problem hiding this comment.
🤖 from Claude (review)
The lease heartbeat fires once per TUPLE, so on any real store the holder's lease expires while it is still writing — the chimera the lease exists to prevent becomes the common case, not a residual race.
on_stage is only invoked from the tuple loop (sweep_stages.py:154-155), so between two heartbeats a whole tuple runs. On the o9/d=4 reference store at the default tuple_width=3, the finest tuple (dispatch=6) is 12·4^6 = 49,152 dispatch nodes, each opening 64 child columns and folding three levels. That is hours against DEFAULT_TTL_S = 900 (sweep_lease.py:49). The lease is therefore expired — and claimable — for essentially the entire first tuple. run_finisher also runs with no heartbeat at all (sweep_stages.py:404).
Confirmed against this branch (lease_ttl_s=2, one dispatch node made to take 3 s):
mid-run sibling admission -> ('SIBLING', 'stage-20260809T080155Z-bd7cb9')
run raised: SweepRefusedError sweep lease ... is no longer held by run
'stage-20260809T080155Z-...' (found 'SIBLING') — another run claimed an
expired heartbeat; aborting
The sibling was admitted with claimed_from naming the live holder, and the holder only noticed at the next tuple boundary — after both had been writing stage artifacts and columns. The _foreign_fresh backstop (sweep_stage.py:289) does not close this: it only fires when a run reads an artifact the other already stamped, so the interleaved PUTs that produce a chimera column happen first.
Repro: run_stage_sweep(root, leaves, lease_ttl_s=2) with any stage_node that sleeps past the TTL, calling acquire_lease(root, run_id="SIBLING") from inside.
The heartbeat needs a cadence bounded by wall time, not by tuple count — e.g. beat from the per-node loop in sweep_stage_pass (or from stage_node's completion) whenever ttl_s/3 has elapsed since the last beat, and beat once more before run_finisher. The per-tuple on_stage row is the wrong clock.
There was a problem hiding this comment.
🤖 from Claude
Folded in 7a994e3: sweep_stage_pass gains an on_node seam fired after every dispatch node, and run_stage_sweep rides it with a wall-clock-throttled beat (ttl_s / 3 since the last beat), keeping the per-tuple beat as well and beating once more immediately before the finisher's RMW. Pinned by TestHeartbeatCadence (the seam fires once per node; the orchestrator beats inside a tuple under a 1 s TTL).
| group = zarr.open_group(self._store, path=str(res), mode="r", zarr_format=3) | ||
| values = group[name][:] | ||
| except KeyError: | ||
| return None # schema evolution: the field postdates this column |
There was a problem hiding this comment.
🤖 from Claude (review)
A single column short one resolution group — or one transient chunk read error — aborts the entire staged sweep, in direct contradiction of the module's soft-barrier posture.
_ColumnReader.read only guards KeyError, which covers a missing array inside an existing group. A missing resolution group raises zarr.errors.GroupNotFoundError, and nothing between here and run_stage_sweep catches it: _gather_slabs / _merge_slabs / _stage_fold have no try, and stage_node's try wraps only _write_stage_overview (line 1011) and write_stage_column (line 1098) — never the fold.
Confirmed against this branch (fixture store from tests/test_sweep_stage.py, one leaf column's 3/ group removed):
RAISED GroupNotFoundError No group found in store
LocalStore('.../1/1/1/1/all.pyramid.zarr') at path '3'
The whole pass dies; nothing is written, no failed count, no under-coverage record.
Reachable ways to get there:
- Deepening the declaration.
overviews: [4]on an o3 store gives leaf columns carrying{4,3}; re-declaringoverviews: [6]makes the sweep gather at cells 5 (column_resolutionswould now be{6,5,4,3}). Every leaf column written before the change is stamped, committed, and short the5/group — the first sweep after the re-declaration takes the whole run down instead of under-covering the stale leaves and healing as the fleet rewrites them. - Any transient object-store fault on a chunk GET during the fold. At 49k dispatch nodes × 64 columns × fields, one 5xx is a certainty per run.
This is also inconsistent within the module: _readers_for (line 419) deliberately counts an unreadable column as failed and yields None so the fold proceeds — the fold's own reads get none of that treatment. Suggest widening the guard to the group-absent case and wrapping the per-(target, level) fold in the same try that already protects the write, counting failed + source_children.unreadable (which is currently hardcoded to 0 everywhere).
There was a problem hiding this comment.
🤖 from Claude
Folded in f5de4dd, two halves: (1) _ColumnReader.read now treats an absent GROUP like an absent field — (KeyError, FileNotFoundError, GroupNotFoundError) reads None, so a deepened declaration (or a transient member fault) under-fills and completes instead of aborting (TestSoftBarrierReadFaults::test_missing_member_group_fills_never_aborts); (2) source_children.unreadable is real now — _root raises on corrupt root metadata (clean absence still reads absent), _readers_for marks such children with an UNREADABLE sentinel, and both fold kernels count it separately from missing (test_corrupt_column_counts_unreadable_not_missing).
| run_started=run_started, | ||
| store_kwargs=store_kwargs, | ||
| on_stage=lambda row: heartbeat_lease(store_root, lease, store_kwargs=store_kwargs), | ||
| level_actuals=level_actuals, |
There was a problem hiding this comment.
🤖 from Claude (review)
level_actuals is shared across partition passes, so partitions=N multiplies every level's source_children by N in the manifest.
The same dict is threaded into every partition's sweep_stage_pass (line 392), and _accumulate_actuals (sweep_stage.py:1185-1186) adds source_children for both the write path and the skip path (sweep_stage.py:984-992). Coarse dispatch nodes intersect every partition MOC — the test at tests/test_sweep_stage.py:846-847 says so explicitly ("later partitions re-visit them and read current") — so each re-visit accumulates the same artifact's counts again.
Confirmed against this branch, identical fixture store swept twice:
partitions=None: {"2": {"folded": 4, ...}, "1": {"folded": 4, ...}, "0": {"folded": 4, ...}}
partitions=4 : {"2": {"folded": 16,...}, "1": {"folded": 16,...}, "0": {"folded": 16,...}}
4 leaves, 4 children folded — the partitions=4 run records 16. run_finisher writes that straight into pyramid.overviews[k].actuals.source_children, which is the #381 point (7) provenance record readers are told to trust. Nothing in the battery pins it: test_partitions_compose_under_one_lease never inspects actuals, and test_per_entry_actuals_land_in_the_manifest only asserts missing == 0.
Same root cause, second symptom: a scoped run (including the default stage_sweep_after_run chaining, which scopes to the fleet's footprint) accumulates only its own subtree, then run_finisher overwrites the previous whole-store actuals with it. After one small append the manifest claims level 0 folded 2 children when the store has thousands.
Both want the accumulator keyed by artifact rather than summed blindly — e.g. level_actuals[k]["source_children"] accumulated per (target, key) seen once, and reset (or merged, not added) per partition pass. Worth a test: partitions=4 and partitions=None over the same fixture must produce identical actuals.
There was a problem hiding this comment.
🤖 from Claude
Folded in bebfc70: _accumulate_actuals is keyed per (artifact node, window) and ASSIGNS rows; aggregate_actuals sums them exactly once at the end, so partition re-visits of shared coarse ancestors (and current-skip re-visits generally) cannot inflate the counts. The partitions test now pins the manifest actuals byte-for-byte against the unpartitioned expectation (folded: 4 at node 0). On the scoped-run half: a scoped run's actuals deliberately record that run's observation — run_id names the writing run, and the spec text frames source_children as "the run's per-artifact coverage counts" — so a scoped update overwriting the previous whole-store row with its own (skip-included: current entries also contribute their recorded rows, which carry the prior coverage forward for every artifact the run visits) is the designed posture. If a cross-run accumulation semantic is wanted instead, that is an espg call — flagged in the PR body.
| if _same_stamp(before, after): | ||
| return values | ||
| logger.info(f"stage sweep: column {self.path} moved mid-read; re-reading") | ||
| self.stamp, self.attrs = after, attrs |
There was a problem hiding this comment.
🤖 from Claude (review)
The stamp-validation re-read drops the foreign-fresh check, so the one race the backstop was ruled for slips through it.
_foreign_fresh is evaluated exactly once, in _ColumnReader.__init__ (line 328), against the stamp read at construction. When the pre/post check here detects movement, the reader adopts the new stamp (self.stamp, self.attrs = after, attrs) and retries without re-testing it. So:
- fleet rewrites the column mid-read → correct, re-read, fold the coherent column (the ruled fleet ∥ sweep case);
- a foreign sweep rewrites the column mid-read → also just re-read, and the merge silently consumes another run's stage output.
The second case is precisely the residual the ruling names ("TTL-expiry clock skew, zombie workers from a crashed run", issue #384 §7 / the phase-2 addition (a)), and given the heartbeat cadence problem I've flagged on sweep_stages.py:391 it is not hypothetical: the zombie is the previous holder, still writing after a sibling claimed its expired lease.
test_stamp_validation_rereads_a_moved_column (tests/test_sweep_stage.py:519) only exercises the fleet arm — it rewrites the leaf with _write_leaf, which stamps no run_id, so _foreign_fresh would return False even if it were called.
One line: re-run the _foreign_fresh(after, ...) test after adopting the moved stamp and raise ForeignSweepError there, same as construction. That also needs the reader to keep run_id/run_started as instance state (currently they are constructor locals).
There was a problem hiding this comment.
🤖 from Claude
Folded in f5de4dd (guard) + 7d5536c (regression test): the reader keeps its run identity, _foreign_guard is re-run on every post-read stamp before a re-read is accepted, and TestForeignStampAtReread drives the exact scenario — a mid-read rewrite stamped by a foreign run — to the loud abort.
| "spec": PYRAMID_SPEC_V2, | ||
| "overviews": expand_overviews([4], parent_order=3), | ||
| "overview": { | ||
| "all_time": False, |
There was a problem hiding this comment.
🤖 from Claude (review)
Test gap: the entire windowed / all-time arm of the stage worker is unexercised — all_time is never True in any test in the repo.
Every staged-sweep test builds its store through _stage_store, whose manifest has no temporal block, so sweep_stage_pass computes windowed = False and _window_work yields the single (SCHEDULE_NONE_TOKEN, [None]) item. all_time is consequently False on every stage_node call in the suite (grep -rn "temporal" tests/test_sweep_stage.py tests/test_sweep_lease.py → nothing). What that leaves untested:
_stage_fold'sres_src = r if (all_time and regime == STAGE_GATHER) else shard_orderbranch (line 599) — the arm that decides whether the all-time fold stays on the gen-1 tier, i.e. the merge-source law's one non-obvious case;_merge_slabs' multi-window path:windows = next(...)(line 501), the exact arm'scombine_denseaccumulation across windows (line 521), and the digest arm folding every window's payloads into onependingcell (line 536);- the
if dispatch == 0 or (windowed and all_time): returnguard (line 1067) — the rule that stops an all-time column from relaying merged content; - per-window stage columns (
window=fold_windows[0], line 1109) and the per-window_readers_forfan-out.
I ran a windowed variant of the fixture by hand (o3/d=1, two windows, all_time: True) and it does behave correctly — 1/2020.zarr = 816, 1/2021.zarr = 2448, 1/all.zarr = 3264, root regime stage-merge, merges_from_raw: 2 — so this is a coverage gap, not a live defect. But it is a large, ruling-governed branch with zero pinning, and the acceptance battery in the PR body makes no claim about it either way. A windowed fixture asserting (a) the all-time artifact equals the cross-window fold of the per-window ones, (b) no all-time stage column is written, and (c) byte-identity across tuple_width with windows, would close it.
There was a problem hiding this comment.
🤖 from Claude
Folded in 663eafd: TestWindowedStageSweep builds a two-window morton-hive/2 store with all_time: true and pins the per-window artifacts (gather regime, gen-1), the all-time fold as stage-merge at exactly 2 merges from raw with the exact-class cross-window sum, and the absence of any all-time stage column at width 1 (per-window columns carry the gen-1 tier).
| # refuses exactly like a live lease. | ||
| prior = (existing or {}).get("run_id") | ||
| try: | ||
| obstore.delete(store, LEASE_NAME) |
There was a problem hiding this comment.
🤖 from Claude (review)
The crash-recovery claim deletes the intent unconditionally, so a second claimant can delete a claim that has already been verified — admitting two holders at once.
The docstring says the verify "closes most of the window", but the delete is what opens the uncloseable part. Interleaving, both runners seeing the same expired intent:
A: delete(expired) A: put(create) -> ok A: verify -> A A: returns, starts sweeping
B: delete(A's FRESH intent) B: put(create) -> ok B: verify -> B B: returns, starts sweeping
A verified before B's delete and never looks again until its first heartbeat — one tuple later (see my note on sweep_stages.py:391). Both are writing stage columns into the same store in the meantime. Note this is not the same as the acknowledged "lost the claim race" path: the loser there refuses, whereas here the earlier claimant wins the verify and is then silently dispossessed.
Cheap tightening: re-read immediately before the delete and skip the delete if the intent is no longer the expired one we read (heartbeat_at moved or run_id changed), then fall through to the ordinary live-lease refusal. That makes the sequence read → confirm-still-expired → delete → create → verify, which turns the double-admission into a refusal for whichever runner arrives second. It does not make the pair atomic, but it removes the case where a fresh lease is deleted.
There was a problem hiding this comment.
🤖 from Claude
Folded in fd068e5: the claim path re-reads immediately before the delete and refuses on ANY movement since the expiry read, so a sibling's already-verified fresh claim can no longer be deleted. The remaining GET→DELETE window is the documented residual the foreign-fresh-stamp abort backstops. Pinned by test_claim_refuses_if_the_intent_moved_before_the_delete (the racing sibling's intent survives intact).
|
🤖 from Claude Three of the "Questions for review" now have follow-up homes (espg-directed): question 6 (same-second skip-gate blind spot) → #417, carrying the cheap fix this PR accidentally enabled — the skip key comparing |
|
🤖 from Claude espg rulings recorded (2026-08-09), closing this PR's review questions: (1) confirmed — family |
1 participant
Closes #384. Refs #381 (umbrella), #383/PR #391 (leaf columns), PR #380 (partition machinery), PR #389 (
/2grammar), PR #397 (lifecycle seam + run-record shape).Implements the staged dense sweep for
zagg-pyramid/2stores per the implementation plan on #384 as amended by the two plan deltas and governed by the four rulings recorded there (merge-source law / finisher / stale-MOC discovery, sweep-admission lease + concurrency matrix):4^widthimmediate child columns; its own column relays the subtree's gen-1 leaf node-order partials as a pure gather; every merge at every level consumes only the relayed gen-1 tier — never a merged tier — so the merge tree is grouping-independent andtuple_width=1/tuple_width=3builds are byte-identical (the single test that pins the law).coverage.mocrefresh, manifest RMW with per-entry actuals, lease release) belong to one designated finisher fired after the root tuple, never to the 12 base cells.coverage.mocis an accelerator only.If-None-Match: *), run id + scope + heartbeat + TTL; live intent refuses naming the runner; expired heartbeat is claimable; stamps carry the run id and a foreign-fresh stamp aborts loudly (the residual-race backstop). Control-plane framing: no data object is ever locked.Phases
src/zagg/sweep_stage.py, pure functions): tuple grouping (tuple_widthknob, dispatch at nodes ≡ 0 mod width, ragged finest tuple), derived gather/merge classification (gather iffcells >= shard_order), column-member derivation (relay + coarser tuples' gatherables), scope normalization (node-prefix MOC; shardmap keys as sugar), scope ∩ partition composition via mortie's scalarmoc_and(TODO markers reference Batch MOC set operations: mocs_and / mocs_intersect and the 1xN broadcast family espg/mortie#173 / PR 174 for the batchmocs_andswap at mortie 0.9.6 — no unreleased-mortie dependency).4^widthchild columns with optimistic stamp validation (pre/post-read check, re-read on movement), level slabs + the worker's own column (relay as pure gather),/2per-artifact attrs (zagg-overview/2: regime, merges-from-raw,source_children), stamps carrying the run id, skip-if-current keyed on summed child generations, foreign-fresh-stamp abort, soft barriers (partial tuples under-cover loudly, self-heal).coverage.mocrefresh sequenced before any scoped fan-out reads it, manifest RMW writing per-entry actuals intopyramid.overviewsentries (composing with, not duplicating, the PR leaf skip-if-current: input-identity no-op with lifecycle touch (issue #388) #397 root-touch — the RMW refreshes the manifest'sLastModified;aggregation.yamlgets the lifecycle touch), the/2default flip inbuild_pyramid_block(+ worker-gate symmetry + fixture regen), lease release as the final act.run_stage_sweep(in-process), the sweep-admission lease (src/zagg/sweep_lease.py),partitions=composition, CLI backstop (python -m zagg.sweep --stages), post-fleet chaining opt-in (output.sweep: "stages"), per-stage + lease intent/completion rows extending the leaf skip-if-current: input-identity no-op with lifecycle touch (issue #388) #397 run-record shape, listing-based unscoped discovery./2per-artifact attrs + stamprun_id(additive), the lease object's byte shape, the retrofit-regime statement (overview sweep: cascade coarse levels from the previous overview (issue #376) #379's cascade remains the/1/pre-column path), the concurrency matrix (fleet∥fleet / fleet∥sweep / sweep∥sweep) + control-plane framing in the operator docs, the sweep-written ancestor-overview aging note (owned here per plan delta 2), conformance fixtures through the generator, raster stores column-less by construction with a clean seam for raster hive under a /2 pyramid declaration: leaf columns are never written (rule PR #391 options (a)/(b)) #399.Module layout
The run machinery split at its natural seams to respect the §4 module cap:
src/zagg/sweep_stage.py(planner + one stage worker's fold kernels and writers, ~1,185 lines),src/zagg/sweep_stages.py(the run: scope planning, pass driver, designated finisher, orchestration, record, chaining, ~585 lines),src/zagg/sweep_lease.py(admission, ~220 lines). After the review folds pushed the worker module past 1,200, the four pure scope/partition-composition helpers moved to the run module (their only consumer) — disclosed here rather than raised as a blocking question since both files sit comfortably under the cap; veto welcome. Tests:tests/test_sweep_stage.py(planner/worker/finisher/orchestration battery),tests/test_sweep_lease.py.Acceptance battery (lands with phases 2–4)
Disjointness pinned per tuple via the #380 spy pattern; byte-identity
tuple_width=1vs=3; gather levels carry gen-1 content untouched; per-entry merges-from-raw == 2 for every upfront merge level (gathers record 1; 3 is append-later cascade territory only); partial-tuple under-coverage loud + self-healing end-to-end; scoped NEON-adjacent update (ancestor prefixes only, old neighbors fold in, clean nodes no-op); stale-root-MOC discovery; lease contention (refusal names the runner), expired-heartbeat claim completing a partial prior run, foreign-stamp abort, stamp-validation re-read, fleet-append-during-live-sweep;/2fully sweepable with the declared-not-yet-sweepable gate retired and/1unchanged.Key mechanics for review
TestMergeSourceLaw::test_byte_identity_across_tuple_widths): builds the same fixture store twice, sweeps one attuple_width=1and one attuple_width=3, and compares every ladder artifact'smorton/exact/digest array contents element-for-element (digest payload bytes compared exactly). It bites because a width-1 build merges through TWO relay hops (stage columns at orders 2 and 1) while width-3 merges straight from the leaf columns — under rejected reading (B) the digest bytes would differ; under Rule A they cannot. A companion spy test (test_merge_consumes_only_the_gen1_tier) asserts stage columns are only ever read at the relay resolution.claimed_from: A— and finishes the rest), foreign-fresh-stamp abort via an injected zombie stamp, stamp-validation re-read (a reader hook rewrites the leaf column mid-read; the pre/post stamp check catches it and the merge consumes the coherent post-rewrite column), and fleet-append-during-live-sweep (missing candidate column recorded as under-coverage, healed by the next pass).Adversarial review (folded)
The fresh-context review posted six inline findings; all six are folded with a reply + fix sha on each thread: per-node lease heartbeat (7a994e3), missing-member tolerance + real
unreadablecounting (f5de4dd), foreign-fresh guard re-run at the validation re-read (f5de4dd + 7d5536c), per-artifact actuals assignment summed once — no partition inflation (bebfc70), claim-race pre-delete re-verify (fd068e5), and windowed/all-time arm coverage (663eafd). One scoped-actuals semantic (a scoped run'sactualsrecord that run's observation, named byrun_id) is left standing as designed and flagged for espg on the thread.How tested
uv run pytest -v(full suite),uv run ruff check src tests,uv run ruff format --check src tests— per-phase results recorded in the phase commits' CI runs. Baseline note:tests/test_lambda_build.py::TestFunctionBuild::test_function_build_succeedsfails on this machine on a cleanmaincheckout (docker/lambda build environment; pre-existing, unrelated — flagged, not fixed).Questions for review
materializedon/2stores — implemented lean: with per-entry actuals landing inpyramid.overviewsentries, the staged sweep does NOT write the family dict's order-keyedmaterializedon/2stores (one source of truth per fact; the/1-era inventory is preserved verbatim where it exists). Flagging per the plan.RequestResponseforward and thedeployment/aws/handler change stay espg's call (also §1: this PR does not touchdeployment/aws/). The worker-eventstageblock is therefore not yet emitted anywhere; raised as a question, not committed.output.sweep: "stages"; the default stays the families sweep. Flip-to-default is espg's call once operationally proven./2default-flip breadth — the ruled "one line" inbuild_pyramid_blockrequired two symmetries to be correct: the worker column gate (leaf_column_plan) must derive the same default resolution (the grid's resolved chunk order), or a default-flipped store would declare leaf levels no worker writes; and grids with no strictly-interior chunk order (K == 1) have no valid/2default, so they keep the/1fallback. Both spelled out in phase 3; please confirm the K == 1 posture.tuple_width=1build materializes stage columns at every order (each tuple needs its child relay), wheretuple_width=3materializes them at dispatch orders only. The byte-identity test therefore pins every ladder artifact (the reader-facing product, which is what pyramid v2 (umbrella): leaf-push columns, (node, cells) level grammar, staged dense sweep #381 point (6) governs); column placement is cadence, documented as orchestration-never-contract in spec §4.6. Flagged for the record.n_leaves+ max stamp timestamp, 1 s resolution). A child rewritten within the same second with the same leaf count reads current; the/1content-hash backstop cannot apply without folding (which is the cost the skip exists to avoid). Inherited tradeoff, flagged.bench_objectsmodel — the CI-owned.github/scripts/bench_objects.pyobject-count model has no term for the leaf column the default now writes;tests/test_benchmark_objects.py::test_hive_sharded_store_matches_modelpins its audited posture with an explicitpyramid: falseand the model's column term is left for espg (this PR does not touch.github/).tests/data/spec/also surfacedgranules.jsonsidecars (issue leaf skip-if-current: input-identity no-op with lifecycle touch (semantic_hash x granule-id set) #388 machinery) absent from the committed fixtures: pre-existing generator drift, included here since the generator is the source of truth.Pre-existing findings (not fixed here)
tests/test_lambda_build.py::TestFunctionBuild::test_function_build_succeedsfails on this machine on a cleanmaincheckout (lambda build environment) — unrelated.ruff checkflags a pre-existingN818onzagg.registry.UnknownCapability(the CI lint bot runs withoutN); this PR's own exception names carry theErrorsuffix.