stage-worker Lambda transport for the /2 staged dense sweep - #525
Conversation
| shard_order = int(manifest["shard_order"]) | ||
| ladder_entries(manifest.get("pyramid") or {}, shard_order) # loud /2 gate | ||
| by_shard, skipped = _normalize_leaves(leaves, shard_order) | ||
| scope = normalize_scope(nodes) if nodes else None |
There was a problem hiding this comment.
🤖 from Claude (review)
(1) An empty or omitted nodes turns one batch invoke into a whole-tuple sweep — silently, and the record says it swept nothing.
scope = normalize_scope(nodes) if nodes else Nonenormalize_scope refuses an empty MOC by name (sweep_stages.py:68 — "an empty sweep scope selects nothing — pass None for whole-store"); this line routes around that guard and converts the empty list to None, which sweep_stage_pass reads as whole store. The handler feeds it straight from the wire: nodes=block.get("nodes") or [] (lambda_handler.py:1305), so a payload that drops nodes, sends [], or misspells the key gets a full-store tuple sweep.
Measured on the suite's own fixture store (tests/test_sweep_stage.py::_stage_store, o3/width 3):
run_stage_worker(..., dispatch=0, nodes=[])
-> stages[0]["nodes"] == 2, stages[0]["written"] == 7, record["n_nodes"] == 0
-> base cells written: ['-2', '1']
Two things break:
- Single-writer. Phase 2 fans a tuple out over N batch invokes. If the
nodeskey is lost for more than one of them (a dispatcher bug, a truncated payload, a retry built from a partial plan), every such worker writes the same overview objects, envelopes and stage columns concurrently. That is exactly the chimera regimesweep_lease.py's header says the lease cannot cover, because all of these workers share one run id and are admitted as siblings by design (run_stage_workerdocstring, andtest_a_sibling_worker_of_the_same_run_is_admitted). The foreign-fresh-stamp backstop does not fire either — samerun_id. - The record lies.
n_nodes: len(nodes)(line 875) reports the requested count, so the record for that invoke saysn_nodes: 0whilestages[0].nodessays 2 and 7 artifacts landed. The dispatcher's barrier and the finisher both read that record.
The same hole has a second mouth: nodes is never validated to sit at dispatch. scope_admits resolves containment in both directions (its own docstring, sweep_stages.py:106-122), so a node handed at the wrong order over-claims its ancestor's whole subtree:
run_stage_worker(..., dispatch=0, nodes=["111"])
-> stages[0]["nodes"] == 1, written == 4 # the entire base cell "1", not one order-3 node
Suggest: refuse an empty nodes for role="stage" by name (the finisher is the only role that legitimately has none), and assert _decimal_order(n) == dispatch for every entry. Both are cheap and both are the difference between a loud bad payload and a quiet double-write.
There was a problem hiding this comment.
🤖 from Claude
Folded in d3d2ba4. run_stage_worker now validates its node set before it reads or writes anything:
- an empty/omitted
nodesraises by name ("...was handed an empty node set — a stage worker folds ITS share of a tuple; an empty set would sweep the whole store as a same-run sibling of every other worker, which admission cannot catch"), soscope = normalize_scope(nodes)is now unconditional andnormalize_scope's own empty-MOC guard is no longer routed around; - every entry must satisfy
zagg.hive._decimal_order(node) == dispatch; any that do not refuse the WHOLE invoke, naming the offenders and their orders, with the rationale in the message and the docstring:scope_admitsresolves containment in both directions, so an off-order node over-claims its ancestor's subtree, and the resulting double-write is invisible to both the lease and the foreign-fresh stamp (both same-run-permissive by design — that is how siblings are admitted at all).
Both checks run before acquire_lease, so a refused invoke leaves no lease and no artifacts.
Tests (tests/test_sweep_stage_fleet.py): test_an_empty_node_set_refuses_instead_of_sweeping_the_store (also asserts no lease, no 1/all.zarr, no -2/all.zarr), test_a_node_off_the_dispatch_order_refuses_the_whole_invoke (["1", "111"] at dispatch 0 — the good node is not swept either), and the handler-level test_a_stage_event_without_nodes_refuses_by_name, which is the payload shape phase 2 generates (500, error_class == "ValueError", no record at the prefix).
| } | ||
| if any(r.get("root_moc_stale") for r in records): | ||
| summary["root_moc_stale"] = True | ||
| summary["finisher"] = run_finisher( |
There was a problem hiding this comment.
🤖 from Claude (review)
(2) The finisher writes the store-root singletons with no lease admission at all — it will happily finish a run whose lease a foreign sweep has already claimed.
run_stage_sweep acquires the lease at the top and holds it through run_finisher (sweep_stages.py:482-563), heartbeating right before the RMW (_maybe_beat() # the finisher's RMW must not start on a stale beat, line 552). run_stage_finisher calls acquire_lease never and heartbeat_lease never — it only passes a release callback. So the one invoke that touches coverage.moc and does the manifest read-modify-write is the one invoke with no admission check.
The gap is not theoretical: a fleet fan-out over thousands of dispatch nodes can outlive DEFAULT_TTL_S (900 s) between the last stage worker's beat and the finisher invoke — that is precisely the window sweep_lease._expired calls claimable. Reproduced against the fixture store:
# stage workers ran under run "F"; the lease then expired and run
# "foreign-runner" claimed the store
run_stage_finisher(root, leaves, run_id="F", records_from=prefix)
sweep lease ... belongs to run 'foreign-runner', not 'F'; leaving it in place
manifest rewritten under foreign lease: True manifest_updated: True
root_moc written: True
lease released: {'released': False} -> still held by: foreign-runner
summary keys: [... no 'error' ...]
So the finisher does the root-MOC GET-union-PUT and the manifest RMW concurrently with the claimant's own finisher, returns statusCode 200, and the only trace is lease.released: False buried in result — a field nothing raises on and phase 2 has no reason to read. release_lease correctly declines to delete someone else's intent (sweep_lease.py:214-219), which is why this fails silent instead of catastrophically.
Suggest an acquire_lease(store_root, run_id=run_id, ttl_s=..., store_kwargs=...) as the finisher's first act: it is idempotent re-admission when the run still holds its own intent (sweep_lease.py:136-137), and SweepRefusedError naming the claimant when it does not — which the handler already turns into a 500 with error_class. At minimum, lease.released == False must not be reported as a clean finish.
There was a problem hiding this comment.
🤖 from Claude
Folded in 27bc64f. run_stage_finisher now takes the lease as its first act:
acquire_lease(
store_root, run_id=run_id, scope=None,
ttl_s=int(lease_ttl_s or DEFAULT_TTL_S), store_kwargs=store_kwargs,
)Idempotent re-admission when the run still holds its own intent; SweepRefusedError naming the claimant when a live foreign one does — which the handler already turns into a 500 with error_class. Added lease_ttl_s to the signature and threaded block.get("lease_ttl_s") through the handler's finisher call so both roles read the same wire key. run_finisher's release stays the final act, and the docstring now records the run_stage_sweep posture explicitly: on any failure the lease is left HELD (nothing on the failure paths releases it), so a half-finished store expires into claimability rather than admitting a sibling.
Tests: test_the_finisher_refuses_under_a_foreign_lease (release F, claim as foreign-runner, finisher raises SweepRefusedError; manifest byte-unchanged and no finisher.json) and test_the_finisher_re_admits_its_own_run (the idempotent half — F's own live intent is re-read, then released).
| by_shard, skipped = _normalize_leaves(leaves, shard_order) | ||
| merged: dict = {} | ||
| stage_rows: list = [] | ||
| records = read_stage_records(records_from, store_kwargs=store_kwargs) if records_from else [] |
There was a problem hiding this comment.
🤖 from Claude (review)
(3) A finisher with no readable records silently records no actuals and still reports success — the exact silent under-reporting _put_stage_record's not-fail-open posture exists to prevent.
records = read_stage_records(records_from, store_kwargs=store_kwargs) if records_from else []records_from is optional here and optional on the wire (block.get("records_from"), lambda_handler.py:1295). When it is absent — or present but listing nothing (wrong prefix, a status bucket the finisher's role cannot list, every stage invoke lost) — merged stays {}, aggregated stays {}, and run_finisher then skips both manifest branches: if node == shard_order and level_actuals: is falsy, elif node in level_actuals never matches, and if changed or level_actuals: gates the PUT off entirely (sweep_stages.py:357-379).
Measured, on a store whose stage workers did run:
run_stage_finisher(root, leaves, run_id="H") # records_from omitted
stage_records: 0 levels: {} manifest_updated: False lease: {'released': True}
actuals after finish: [None, None, None, None]
Root MOC refreshed, aggregation.yaml touched, lease released, run record written, HTTP 200 — and the #381 point (7) per-level actuals, the entire reason the stage records exist as an aggregation channel, are simply absent. Every downstream reader sees a completed /2 sweep with no provenance.
This is inconsistent with the PR's own argument in "Questions for review" (1): _put_stage_record raises rather than fail open because "swallowing a failed PUT would silently under-report coverage in the one place #381 point (7) exists to record it". The read half of the same channel does exactly that swallow. Suggest: make records_from required for role="finisher", and treat len(records) == 0 (or, better, a record count that misses the dispatcher's declared batch names) as a raise, not a 200.
There was a problem hiding this comment.
🤖 from Claude
Folded in b47c06f. Two refusals, both by name:
records_fromis required — absent raises"the finisher for run ... was handed no records_from — the run's stage records ARE its per-level actuals; finishing without them would stamp the manifest with no coverage at all". The trailingif records_from:guard on thefinisher.jsonPUT is gone with it.- zero readable records raises
"...found no stage records under <prefix> — a finisher fires only after a fan-out, so zero records means the fan-out was lost; refusing rather than recording an empty sweep".
A PARTIAL set stays fine, as you say — recorded-and-healed under-coverage (#381 point (6)); only zero refuses, since only zero is indistinguishable from "nothing ran". I did not go as far as checking the record names against a dispatcher-declared batch list: the dispatcher does not ship that list today, so it would be a new event-grammar key. Leaving that standing as a phase 2 question rather than inventing the key here.
Tests: test_a_finisher_without_records_from_refuses, test_a_finisher_over_zero_records_refuses (manifest byte-unchanged, no finisher.json, and the lease still held by F — the failure posture from finding (2)), and test_a_partial_record_set_still_finishes (one of two records deleted → finishes, folded == 3 where both would give 4: under-reported, loudly, not refused).
| from zagg.store import open_object_store | ||
|
|
||
| store = open_object_store(records_from, **dict(store_kwargs or {})) | ||
| listing = obstore.list_with_delimiter(store) |
There was a problem hiding this comment.
🤖 from Claude (review)
(4) read_stage_records never filters by run_id, so a stale record from an earlier attempt under the same prefix can win the merge.
listing = obstore.list_with_delimiter(store)
keys = sorted(... startswith("stage-") ... endswith(".json"))The only filters are the basename prefix and the spec check. Every record already carries run_id and run_started (sweep_stages.py:866-868) and run_stage_finisher has the run id in hand — it just does not use it.
The prefix is run-id-scoped (run_status_prefix → <store>.status/run-<run_id>), so this is safe only while a run id is used exactly once. It is not safe for the case the PR explicitly contemplates — a retried/resumed run under the same id — because the object names are stage-<dispatch>-<batch>.json and the batching is a dispatcher choice, not a store fact. Resume with fewer or differently-numbered batches (4 batches → 2) and stage-00-0002.json / stage-00-0003.json from the dead attempt survive. merge_level_actuals then folds them, and since it assigns per (node|window) in sorted-key order, the stale higher-batch record's rows overwrite the fresh lower-batch ones — stale counts win, permanently, in the manifest.
test_merge_is_idempotent_over_reinvoked_batches does not cover this: it re-fires the same batch numbers, which overwrite their own objects.
One-line fix: if record.get("run_id") == run_id in the loop (thread run_id through), or have the finisher be handed the exact record names the dispatcher declared and read those, which also lets it detect a missing one (see finding (3)).
There was a problem hiding this comment.
🤖 from Claude
Folded in 6e82689. read_stage_records now takes run_id as a required keyword and skips (with a warning naming the foreign run) any record whose run_id does not match; run_stage_finisher threads its own id in. The docstring states the reason: the prefix is run-scoped but the object NAMES are stage-<dispatch>-<batch> and the batching is a dispatcher choice, so a resumed run under the same id leaves the dead attempt's higher-numbered records in place, and merge_level_actuals ASSIGNS in sorted-name order — stale rows would win, permanently.
The regression test you flagged as missing: test_a_prior_attempts_record_cannot_win_the_merge — run F writes stage-00-0000/stage-00-0001, then a copy of batch 0 with run_id: "E" and every folded set to 99 is written as stage-00-0009 (sorts LAST, so it would win an unfiltered merge). The finisher reports stage_records == 2 and levels["0"]["source_children"]["folded"] == 4, not 99. Plus a direct unit test, test_read_stage_records_filters_on_run_id, pinning both directions of the filter.
| "duration_s": time.perf_counter() - t0, | ||
| "discover_s": discover_s, | ||
| "record": out.get("record"), | ||
| "result": out, |
There was a problem hiding this comment.
🤖 from Claude (review)
(5) The response embeds the whole record twice and can blow the 6 MB invoke-response cap on work that actually succeeded — and record means two different things depending on role.
"record": out.get("record"),
"result": out,Size. For role="stage", out is the stage record, level_actuals and all — one row per (artifact node, window). Measured on the real serialization (json.dumps(..., indent=1) as _put_stage_record uses):
1000 artifact rows -> 92,130 bytes (~92 B/row)
A finest-tuple batch of 200 order-6 dispatch nodes covers orders 8/7/6 = 21 artifact nodes each = 4,200; times 12 monthly windows = ~50k rows ≈ 4.6 MB, plus the stages rows and the duplicated record URL. Lambda's synchronous response limit is 6 MB, so a batch that folded correctly and PUT its record correctly comes back to a RequestResponse driver as ResponseSizeTooLarge — indistinguishable, from the dispatcher's side, from the invoke failure the PR body says a 500 exists to disambiguate. The record is already durable at the status prefix and the finisher reads it from there; the response does not need to carry it. Suggest returning the record's identity (dispatch/batch/URL) plus the stages rows and dropping result, or at least dropping level_actuals from it.
Shape. body["record"] is a full URL (<records_from>/stage-00-0002.json) for role="stage" but a bare store-root key (sweep_stats_…_stages.json) for role="finisher", which is also what the families arm puts under that key (_handle_sweep, summary.get("record")). Three meanings, one key, one mode: "sweep" envelope. The finisher's status-prefix record is meanwhile only reachable as result.finisher_record. A driver cannot key off body["record"] at all. Suggest distinct keys (stage_record vs run_record) or a role-independent contract, settled now while phase 2 is still being written against it.
There was a problem hiding this comment.
🤖 from Claude
Folded in 0357fe8. New _stage_body() next to _handle_stage_sweep builds the envelope for both roles and both outcomes; "result": out is gone.
Size. No record, no rows, no level_actuals cross the wire — only scalars: ok, mode, stage, run_id, dispatch, batch, n_nodes, n_leaves, stage_records, lease_released, written/current/failed/under_covered (summed over the invoke's stage rows), duration_s, discover_s, record, stage_record, error, error_class.
Shape. record is now unambiguously the store-root run record (sweep_stats_*_stages.json) — the finisher's only, None for a stage invoke, matching what the families arm puts under that key. The status-prefix object gets its own key, stage_record: the stage record's URL for role="stage", finisher.json's for role="finisher". The key SET is identical across both roles and both outcomes (None where a role has no such value), so a driver never KeyErrors — including on the 500 path, which now carries the same keys plus ok: false.
lease_released is the one addition beyond your list: it keeps finding (2)'s "must not be reported as a clean finish" visible to a driver as a top-level scalar rather than buried in result.
Tests: test_the_envelope_is_scalars_and_one_fixed_key_set (asserts set(stage) == set(finisher) == set(failed), that no value is a dict or list, and that level_actuals/result are absent), plus the two round-trip tests updated to read the new keys.
| dispatch=int(block["dispatch"]), | ||
| nodes=block.get("nodes") or [], | ||
| batch=int(block.get("batch", 0)), | ||
| tuple_width=int(block.get("tuple_width", 3)), |
There was a problem hiding this comment.
🤖 from Claude (review)
(6) The wire default for tuple_width is hardcoded 3 instead of DEFAULT_TUPLE_WIDTH.
tuple_width=int(block.get("tuple_width", 3)),zagg.sweep_stage.DEFAULT_TUPLE_WIDTH is the single source everywhere else — sweep_stage_pass, run_stage_sweep, run_stage_worker all take it as their default. This arm copies the value, and the module docstring restates it a second time ("tuple_width": int, # optional, default 3).
stage_tuples' grouping determines which orders are dispatch orders, hence which nodes get a stage column written at all. If DEFAULT_TUPLE_WIDTH ever moves, a dispatcher that omits tuple_width gets width-3 tuples from the handler while python -m zagg.sweep --stages gets the new width — different column sets on the same store, from the same store state. That is precisely the grouping-independence the merge-source law is being leaned on for, undone by a literal. The function already does from zagg.sweep_stages import ... locally; add DEFAULT_TUPLE_WIDTH to it (or better, require the key: the dispatcher pins it anyway, per the PR's own event grammar).
There was a problem hiding this comment.
🤖 from Claude
Folded in 1597f7b. The arm now imports DEFAULT_TUPLE_WIDTH from zagg.sweep_stage alongside its other local imports and uses int(block.get("tuple_width", DEFAULT_TUPLE_WIDTH)). The module docstring's second copy of the literal is gone too — it now names the constant and says why (a copied literal would give the fleet a different tuple grouping, hence a different set of stage columns, on the same store).
I left the key optional rather than requiring it: making it mandatory would break the "tuple_width": int, # optional grammar the PR body already published, and that is a scope call for the dispatcher work. Leaving that half standing.
Test: test_the_wire_default_tuple_width_is_the_single_source — a stage event with tuple_width deleted, with run_stage_worker spied via monkeypatch, asserts the handler passed sweep_stage.DEFAULT_TUPLE_WIDTH rather than a literal.
|
|
||
| for stage in stage_tuples(shard_order, tuple_width=tuple_width): | ||
| schedule = stage_tuples(shard_order, tuple_width=tuple_width) | ||
| if only_dispatch is not None: |
There was a problem hiding this comment.
🤖 from Claude (review)
(7) The only_dispatch refusal sits behind two early returns, so the no-op it promises to prevent is still reachable.
The new docstring (line 160-162) states the contract plainly:
An order that dispatches no tuple refuses BY NAME rather than sweeping nothing — a mistyped dispatch order must not read as a clean no-op.
But the check runs at line 204, after if not fields: ... return summary (line 189) and if not candidates: return summary (line 196). A stage invoke against a store with no composable fields declared, or one handed a leaf slice that is empty with no usable root coverage.moc, returns {"stages": []} for any only_dispatch value — including a garbage one — and run_stage_worker then PUTs a perfectly well-formed record with stages: []. The dispatcher's barrier sees the record and moves on; the mistyped order never surfaces.
Validating only_dispatch against stage_tuples(shard_order, tuple_width=...) right after shard_order is read (before the fields/candidates gates) makes the guard match its docstring; it is a pure function of the two arguments and needs nothing from the store.
There was a problem hiding this comment.
🤖 from Claude
Folded in 19eba57. schedule = stage_tuples(shard_order, tuple_width=tuple_width) and the only_dispatch refusal now run immediately after shard_order/cell_order/ladder_entries — above the decl/fields gate and above _candidate_decimals, so both early returns are behind it. A comment at the site records why: it is a pure function of shard_order and tuple_width and needs nothing on disk, and a mistyped order against a store with nothing to sweep would otherwise return stages: [] and PUT a well-formed record the dispatcher's barrier accepts.
Test: test_a_mistyped_dispatch_refuses_even_with_nothing_to_sweep — a manifest copy with pyramid.overview.fields = {} (so the not fields gate is live) asserts the ordinary pass still returns stages == [], then that only_dispatch=7 raises "no stage tuple dispatches at order 7" on the same store; and the same again through run_stage_worker with an order-7 node set, which is the real fleet path.
| } | ||
| }, | ||
| ) | ||
| assert target[2]["children"] == {"111|all": {"folded": 2, "missing": 0, "unreadable": 0}} |
There was a problem hiding this comment.
🤖 from Claude (review)
(8) test_merge_level_actuals_first_wins_on_level_metadata asserts nothing about first-wins level metadata.
Both merge_level_actuals calls pass identical cells/regime/merges_from_raw (3 / "stage-gather" / 1), and the single assertion only inspects target[2]["children"]. Every level-metadata value would pass unchanged if the setdefault at sweep_stages.py:695 were replaced with a plain assignment (last-wins) — the test cannot fail for the reason it is named after.
That matters because first-wins across workers is load-bearing for the merge-source law: the merge order is sorted record-name order (stage-<dispatch>-<batch>), which is a dispatcher batching artifact, so if the values ever disagreed the manifest's recorded regime would depend on grouping. The docstring argues they cannot disagree (both derived per level via classify_level) — pin that: have the second call carry a different regime/merges_from_raw and assert the first one survives.
Two smaller ones in the same file:
- Line 15: the module docstring says "see
TestByteIdentityOracle" — no such class exists here or anywhere intests/(grep -rn TestByteIdentityOracle tests/returns only this line). It is phase 3 work; the docstring reads today as if the acceptance already lands in this PR. - There is no negative test for finding (1) — nothing pins that a
stageinvoke withnodesmissing or[]refuses, which is the payload shape phase 2 is about to start generating.
There was a problem hiding this comment.
🤖 from Claude
Folded in 287eab9 (all three points).
The merge test. The second merge_level_actuals call now disagrees on all three level-metadata values — cells: 999, regime: "stage-merge", merges_from_raw: 7 — and the test asserts the first ones survive (3 / "stage-gather" / 1), plus that the per-(node, window) rows still merge in from both. Replacing the setdefault at sweep_stages.py with a plain assignment now fails it, which was the point. The comment says why the tie is pinned even though it cannot legitimately arise: merge order is sorted record-NAME order, a dispatcher batching artifact, and the merge-source law does not let the recorded regime depend on grouping.
The dangling name. Kept the reference but dated it — the module docstring now reads "its TestByteIdentityOracle lands with phase 3 of this PR and is NOT in this file yet; phases 1-2 pin the transport", so it no longer reads as an acceptance that already ships.
The finding (1) negative test. Added, as test_an_empty_node_set_refuses_instead_of_sweeping_the_store and the handler-level test_a_stage_event_without_nodes_refuses_by_name — see the reply on that thread (fix d3d2ba4).
| cur_leaves: list = [] | ||
| cur_bytes = 0 | ||
| for node in nodes: | ||
| refs = _leaf_refs(by_shard, [node]) |
There was a problem hiding this comment.
🤖 from Claude (review)
pack_batches is O(nodes × leaves) — the quadratic term the docstring says it avoids is still here.
for node in nodes:
refs = _leaf_refs(by_shard, [node])_leaf_refs (line 76) does sorted(by_shard) and a startswith over the entire work set on every call, so the loop is len(nodes) full sorts + full scans. The docstring at line 124 says the incremental byte accounting exists because "a full json.dumps per candidate node would be quadratic on the ~49k-node finest tuple of an o9 store" — that fixed the dumps cost and left the dominant term untouched.
Measured on this branch (.venv/bin/python, one base cell, nodes = leaves truncated three orders up), each step is 4× the nodes and 4× the leaves:
leaves=16384 nodes=64 batches=2 pack_s=0.33
leaves=65536 nodes=256 batches=8 pack_s=1.19
leaves=262144 nodes=1024 batches=30 pack_s=17.75
0.33 → 1.19 → 17.75 is quadratic, not linear. Extrapolating the same shape to the case the docstring names — an o9 store, ~49k dispatch nodes at order 6, millions of shard decimals — this is tens of minutes to hours of pure dispatcher CPU before the first invoke fires, on top of the barrier wall clock, and it happens once per tuple.
The fix is one pass instead of len(nodes) passes: bucket the work set by _node_at(d, dispatch) once, then index. Something like
buckets: dict[str, list] = {}
for decimal in sorted(by_shard):
buckets.setdefault(_node_at(decimal, dispatch), []).extend(...)and have pack_batches take (or build) that map, with _leaf_refs(by_shard, [node]) reserved for the finisher's whole-set call at line 360.
There was a problem hiding this comment.
🤖 from Claude
Fixed in 3dc0c8b. pack_batches now buckets the work set by zagg.sweep_stage._node_at ONCE (_bucket_leaf_refs, src/zagg/sweep_fleet.py) and looks each node up in the map, instead of calling _leaf_refs(by_shard, [node]) per node. _leaf_refs stays for the finisher's whole-set call.
Bucket slices are byte-identical to the per-node prefix scan — pinned by TestBatching::test_the_node_buckets_match_the_per_node_slices, which walks all four dispatch orders on a two-window work set and asserts buckets[node] == _leaf_refs(by_shard, [node]), order included.
| cur_bytes = 0 | ||
| for node in nodes: | ||
| refs = _leaf_refs(by_shard, [node]) | ||
| cost = len(node) + 4 + sum(len(json.dumps(r)) + 1 for r in refs) |
There was a problem hiding this comment.
🤖 from Claude (review)
The incremental accounting undercounts by one byte per leaf ref, so a batch pack_batches believes is inline actually ships as discover: true — silently.
cost = len(node) + 4 + sum(len(json.dumps(r)) + 1 for r in refs)json.dumps uses the default separators (", ", ": "), so each element after the first in the leaves list costs len(json.dumps(r)) + 2 (comma and space), not + 1. The node term is right ("n", = len+4); the leaf term is off by one, and it is the term that repeats thousands of times. base carries only a fixed + 64 margin (line 137), which cannot absorb ~1 byte × 9,000 refs.
This is not theoretical — it reproduces on the fixture this PR's own test uses (generate_morton_children(morton_word("1"), 7), 16,384 leaves, nodes d[:4]), run against pack_batches on this commit:
CAP 256000 nodes 64 leaves 16384 batches 2
0 nodes 35 refs 8960 bytes 260369 OVER
1 nodes 29 refs 7424 bytes 215777
Batch 0 projects to 260,369 bytes — 4,369 over _ASYNC_PAYLOAD_CAP_BYTES. build_stage_event line 114 then catches it, deletes leaves, and sets discover: true. So the batch that pack_batches sized to ride inline instead makes the worker do a store-wide discover_leaves LIST plus a parquet read — exactly the cost line 96-97 says pack_batches "spends its budget on keeping leaves inline" to avoid. And the conversion happens with no log line at all, so an operator sees only the extra worker latency.
Two things to fix: charge + 2 per ref (or measure the batch with a real json.dumps once per batch, which is linear, not quadratic), and make the build_stage_event fallback at line 114 log a WARNING naming the batch — a silent switch from inline to store-wide discovery is a behaviour change worth one line.
There was a problem hiding this comment.
🤖 from Claude
Fixed in acf99cd. Three changes:
- The estimate now charges the real
", "separator (len(json.dumps(r)) + 2), so it is conservative rather than one byte short per ref. - Careful is not the same as verifiable, so the estimate is no longer trusted on its own: after greedy packing every batch is measured with ONE real
json.dumps(_fit_batch) and halved recursively if it still exceeds the cap. A single node that cannot fit alone is the discover case and is emitted as one. That needed splittingbuild_stage_eventinto_inline_event(no fallback — the shape you measure) plus the capped builder, since measuring through the fallback measures the stripped event and calls every overflow a fit. build_stage_eventlogs a WARNING naming the run, role, batch and dispatch order whenever it does fall back, with the cost spelled out (store-wide LIST plus a parquet read). Pinned byTestBatching::test_the_discovery_fallback_is_logged_by_name.
On your fixture (generate_morton_children(morton_word("1"), 7), 16,384 leaves, nodes d[:4]) the two batches now measure 252,918 and 223,190 bytes — both inline, where batch 0 previously projected 260,369 and shipped as discover: true.
| event = build_stage_event( | ||
| "s3://bucket/p.zarr", {**block, "nodes": batch_nodes, "batch": batch}, refs | ||
| ) | ||
| assert len(json.dumps(event)) <= _ASYNC_PAYLOAD_CAP_BYTES |
There was a problem hiding this comment.
🤖 from Claude (review)
This test cannot fail — it asserts an invariant build_stage_event enforces unconditionally, not the one pack_batches is supposed to hold.
event = build_stage_event(
"s3://bucket/p.zarr", {**block, "nodes": batch_nodes, "batch": batch}, refs
)
assert len(json.dumps(event)) <= _ASYNC_PAYLOAD_CAP_BYTESbuild_stage_event (src/zagg/sweep_fleet.py:113-117) ends with if len(json.dumps(event)) > _ASYNC_PAYLOAD_CAP_BYTES: del event["leaves"]; event["discover"] = True. Whatever pack_batches returns, the event it produces is under the cap by construction — the assertion is a tautology over build_stage_event, and it would still pass if pack_batches were replaced by return [(nodes, _leaf_refs(by_shard))].
And it is masking a live defect right now: on this exact fixture, batch 0 projects to 260,369 bytes (see the pack_batches comment on src/zagg/sweep_fleet.py:145) and only passes here because the fallback stripped its leaves.
To actually test the budget, assert on the payload with the leaves still inline and require the leaves to have survived:
event = build_stage_event(..., refs)
assert refs is None or "leaves" in event, f"batch {batch} overflowed and fell back to discovery"which fails today, as it should.
There was a problem hiding this comment.
🤖 from Claude
Fixed in 4d1d2b5. The test now asserts what pack_batches owes rather than what build_stage_event enforces: for every batch it sized to ride inline, refs is not None, len(event["leaves"]) == len(refs), "discover" not in event, AND the built event is under the cap. It also asserts len(batches) > 1, so the fixture cannot silently collapse to the trivial case.
It fails on the pre-fix accounting exactly as you describe — batch 0's leaves get stripped, so event["leaves"] raises KeyError.
| f"stage fleet: tuple @{dispatch} — {len(nodes)} node(s) in {len(batches)} invoke(s), " | ||
| f"{len(seen)}/{len(expected)} record(s) in {time.perf_counter() - t_stage:.1f}s" | ||
| ) | ||
| _fire( |
There was a problem hiding this comment.
🤖 from Claude (review)
The finisher fires unconditionally, including when zero stage invokes were fired — a guaranteed refusal followed by a full barrier timeout.
The tuple loop continues when a tuple has no dispatch nodes (line 298), and nothing tracks whether ANY invoke went out; _fire for the finisher is unguarded. So when the work set produces no nodes at all, the dispatcher fires exactly one invoke — the finisher — and that invoke is designed to refuse:
if not records:
raise ValueError(
f"the finisher for run {run_id!r} found no stage records under "
f"{records_from} — a finisher fires only after a fan-out, so zero records "
"means the fan-out was lost; refusing rather than recording an empty sweep"
)(src/zagg/sweep_stages.py:1021)
It is an Event invoke, so the 500 is invisible to the dispatcher; await_records at line 364 then blocks the full barrier_timeout_s (1200 s by default) waiting for finisher.json that can never appear, and returns landed: False. On the wired path that is 20 minutes of dead dispatcher wall clock after the fleet already finished.
Reachable ways to get an empty node set with a non-empty leaves argument (the runner tail guards only if leaves:):
- every leaf skipped by
_normalize_leaves(skippedis already recorded in the summary at line 269, so the code knows), - a
scopethat admits no dispatch node —normalize_scoperefuses only an empty MOC, not one naming a subtree with no dirty shards.
Guard it: track whether any batch fired and, if none did, return the summary with finisher: {"landed": False, "skipped": "no dispatch nodes"} without the invoke and without the barrier.
There was a problem hiding this comment.
🤖 from Claude
Fixed in b5b64ed. When no tuple produced a dispatch node the dispatcher returns before the finisher: no invoke, no barrier, one logger.info naming the leaf count and the skip count. The summary records it as "skipped": "no dispatch nodes" with finisher: {"landed": False, "fired": False}.
The key set stays deterministic: "skipped" is initialized to None in the base summary and "fired": True was added to the normal-path finisher block, so a caller never has to know which branch produced the summary.
Pinned by TestFleetOrchestration::test_no_dispatch_nodes_fires_nothing_at_all, which asserts client.events == [], the summary keys, and duration_s < 5 against barrier_timeout_s=30 — i.e. that no barrier was waited on.
| # every stage artifact is regenerable and | ||
| # `python -m zagg.sweep --stages` is the backstop). | ||
| if config.output.get("sweep") == "stages": | ||
| _invoke_lambda_stage_sweep( |
There was a problem hiding this comment.
🤖 from Claude (review)
The seam adds an unbounded, unconfigurable block to the dispatcher tail — up to (n_tuples + 1) × 1200 s, with no total budget and no knob.
_invoke_lambda_stage_sweep(
state["lambda_client"],
function_name,
store_path,
leaves,
shard_order=int(parent_order),_invoke_lambda_stage_sweep never passes barrier_timeout_s, so run_stage_sweep_fleet takes DEFAULT_BARRIER_TIMEOUT_S = 1200 per barrier, and there is one barrier per tuple plus the finisher. For the canonical o9 store at the default tuple_width=3, stage_tuples(9, 3) dispatches at orders 6, 3, 0 → 4 barriers → 4,800 s (80 min) of worst-case blocking, added after the fleet has already landed and before the run's own Done: / cost lines at runner.py:4306. Add the pack_batches CPU cost (separate comment) and the tail can dominate the run.
Three consequences worth a decision rather than a default:
- Every other tail invoke here is fire-and-forget precisely so the dispatcher can exit; this one makes the dispatcher's lifetime a function of fleet queueing. A dispatcher killed mid-barrier (CI timeout, Ctrl-C, a driving Lambda's own 900 s ceiling) leaves the run's lease held — the finisher is the only releaser and it never fires.
- There is no way to tune it: neither
_invoke_lambda_stage_sweepnor the call site exposesbarrier_timeout_s/poll_interval_s, so an operator cannot shorten it for a small store or lengthen it for a throttled account. - The failure mode is silent-ish:
run_stage_sweep_fleetrecordsbarrier_timed_outper stage, but the seam's log line (runner.py:5628) printsrecords_seen/batchesat INFO and never says "a barrier expired". A partially-covered staged sweep reads like a clean one in the run log.
Suggest: thread barrier_timeout_s (and a total budget across tuples) through the seam from config, and log a WARNING when any(s["barrier_timed_out"] for s in summary["stages"]).
There was a problem hiding this comment.
🤖 from Claude
Fixed in abca5be (and 134175e for the WARNING's flag).
- Total budget.
run_stage_sweep_fleettakestotal_barrier_budget_s(DEFAULT_TOTAL_BARRIER_BUDGET_S = 7200) alongside the per-barrierbarrier_timeout_s. Every barrier goes through a_barrier()closure that clamps its budget to what is left of the total and decrements it, so the worst case is a constant instead of(n_tuples + 1) xthe per-barrier budget. Past the total each remaining barrier degrades to a single check — fail-open, exactly as a timeout is.await_recordsalso clamps its sleep to the deadline, so a poll interval longer than the remaining budget cannot overshoot it. - Knobs. Both are keywords on
_invoke_lambda_stage_sweepand forwarded only when set, so the default stays single-sourced inzagg.sweep_fleet. - Docstring. The seam now documents the worst-case wall clock (the total, not the product), that a dispatcher killed mid-barrier leaves the lease HELD because the finisher is its only releaser — and that this is the lease's designed recovery, not a leak, since the TTL expires into claimability and the next sweep takes the store over by name — and that the ladder already written stays valid either way (D9, pyramid v2 (umbrella): leaf-push columns, (node, cells) level grammar, staged dense sweep #381 point (6)).
- Point (3). The seam now logs a WARNING when
summary["barrier_timed_out"], saying the ladder may be partially covered and the finisher's actuals may under-report.
Tests: test_the_total_barrier_budget_bounds_the_whole_tail (four 0.5 s barriers under a 0.5 s total must finish inside 1.5 s), TestRunnerSeam::test_the_seam_threads_the_barrier_knobs, TestRunnerSeam::test_the_seam_says_a_barrier_expired.
|
|
||
| from __future__ import annotations | ||
|
|
||
| import json |
There was a problem hiding this comment.
🤖 from Claude (review)
The barrier budget is sized against the function timeout, but Event invokes queue for far longer than that — a timed-out barrier then lets the finisher release the lease while workers are still in flight.
#: Barrier budget per tuple, in seconds. A stage invoke is bounded by the
#: function timeout (900 s on the deployed fleet), and Event invokes queue, so
#: the budget has to cover a cold fan-out plus one worker's full run.
DEFAULT_BARRIER_TIMEOUT_S = 1200The comment names the right hazard and then sizes for the wrong one. 1200 s covers one worker running; it does not cover the queue. Lambda's asynchronous invocation queue retains events for up to 6 hours, and a fan-out of hundreds-to-thousands of Event invokes against a reserved-concurrency or account-concurrency ceiling is exactly the case where an invoke starts an hour after it was fired. So the reachable state is not "the invoke was lost" (which the soft-barrier reasoning covers) but "the invoke has not started yet".
That distinction matters at line 351: after the last tuple's barrier expires, the dispatcher fires the finisher, which runs run_finisher — root coverage.moc GET-union-PUT, the manifest RMW, and lease release as its final act (src/zagg/sweep_stages.py:286-325). Releasing the lease while queued stage workers of the same run are still pending admits a foreign sweep into a store that still has live writers. It does not corrupt — the straggler and the claimant each hit ForeignSweepError on the other's fresh stamp — but a loud abort of an unrelated later run is a materially worse outcome than the "under-coverage that heals next pass" the docstring at lines 190-196 promises, and it is the exact race the lease exists to prevent.
The await_records docstring says a timeout "is logged loudly and returned, never raised". Worth spelling out the queued-invoke case there and deciding one of: (a) size the default to the async retention the fleet actually runs with, (b) skip the finisher when the final tuple's barrier expired (leave the lease to expire into claimability, which is run_stage_sweep's own posture on failure), or (c) document explicitly that a timed-out barrier can release the lease early.
There was a problem hiding this comment.
🤖 from Claude
Fixed in 134175e, taking your option (c) plus a raised default — the barrier stays soft.
- Default.
DEFAULT_BARRIER_TIMEOUT_S = 2700, and the comment now says what it is sized against: three function timeouts (900 s), one for the fan-out to drain out of the async queue behind an account- or reserved-concurrency ceiling, one for a throttle-driven redelivery, one for the slowest worker's own full run. It explicitly does NOT cover the queue's 6-hour worst case, and says so. - Honesty in the docstring.
await_recordsnow spells out the queued case: an expiry likelier means the invoke has not STARTED; such a worker is a same-run sibling so nothing refuses it (ForeignSweepErroris about other runs, by design) and the column it eventually writes is correct; what the run loses is bookkeeping — the finisher aggregated before that record existed, so its per-level actuals under-report, and its lease release lands while same-run writers are still pending, admitting a later foreign sweep early (each side then aborts on the other's fresh stamp, so it does not corrupt, but it is louder than the under-coverage the soft barrier promises). - Recorded in the outcome.
barrier_timed_outis now a run-level summary key (initializedFalse, so the key set is deterministic) AND rides in the finisher's stage block, so the RUN record says the actuals may be short rather than reading clean. The runner seam warns on it (thread onrunner.py:4293).
Pinned by the extended test_a_lost_invoke_times_the_barrier_out_and_the_run_proceeds (summary flag and the finisher block's flag both True, both False on the healed re-run).
|
|
||
| for stage in stage_tuples(shard_order, tuple_width=tuple_width): | ||
| dispatch = int(stage["dispatch"]) | ||
| nodes = dispatch_nodes(by_shard, dispatch, scope) |
There was a problem hiding this comment.
🤖 from Claude (review)
scope is never normalized here, unlike the in-process twin — a caller using any of the documented spellings gets a mortie error or wrong filtering.
nodes = dispatch_nodes(by_shard, dispatch, scope)run_stage_sweep normalizes first (src/zagg/sweep_stages.py:487):
scope_words = normalize_scope(scope)and normalize_scope's docstring is explicit about what callers may pass: "an iterable of morton words (ints) or D1 decimal strings … or a mapping (a shardmap: its KEYS are the prefixes)". run_stage_sweep_fleet takes the same-named parameter, documents nothing about the required form, and hands it straight to scope_admits, which does moc_and(np.asarray([word], dtype=np.uint64), scope). A list of decimal strings, a shardmap dict, or a plain list of ints will not behave like the in-process path — best case it raises inside mortie, worst case it silently filters differently.
stage_sweep_after_run — the local chaining this module is the twin of — passes a list of decimal strings (src/zagg/sweep_stages.py:641), so the mismatch is with the very caller shape this transport mirrors. The tests only ever pass normalize_scope([...]) (tests/test_sweep_stage_fleet.py:1037), so nothing catches it.
One line: scope = normalize_scope(scope) near the top of run_stage_sweep_fleet, alongside the tuple_width / shard_order coercions at lines 259-261. Also worth recording scope in the returned summary the way run_stage_sweep does ("scope": [...] at sweep_stages.py:514) — the dispatcher summary at line 262 is currently silent about what the run covered.
There was a problem hiding this comment.
🤖 from Claude
Fixed in 3ff57c3. run_stage_sweep_fleet runs scope = normalize_scope(scope) alongside the tuple_width / shard_order coercions, exactly as run_stage_sweep does, with a comment naming stage_sweep_after_run's decimal-string spelling as the caller shape this transport mirrors. The summary also records "scope": None | [str(int(w)), ...], matching sweep_stages.py:514.
New test test_a_raw_decimal_scope_is_normalized_like_the_in_process_pass passes scope=["1111"] raw and asserts the fan-out narrows the same way the pre-normalized call does; test_scope_narrows_the_fan_out now also checks the recorded scope.
|
|
||
| try: | ||
| listing = obstore.list_with_delimiter(open_object_store(records_from, **store_kwargs)) | ||
| except Exception as e: # a prefix with no objects yet, or a transient fault |
There was a problem hiding this comment.
🤖 from Claude (review)
A permanently broken LIST is indistinguishable from an empty prefix, and it is logged at DEBUG — so the whole run burns its full barrier budget in silence.
except Exception as e: # a prefix with no objects yet, or a transient fault
logger.debug(f"stage fleet: cannot list {records_from} ({e})")
return set()The comment names two benign causes, but the bare except Exception also swallows the permanent ones, and those are the likely ones for this specific call: the status prefix is a store sibling (<store>.status/run-<id>, client_transport.py:121-123), not a path under the store root, so a dispatcher role scoped to the store prefix has s3:ListBucket on the store and not on .status. Same for a wrong region, a wrong endpoint in store_kwargs, or expired credentials.
In every one of those cases _present returns set() forever, await_records spins to the deadline on each tuple, and the only WARNING the operator gets is the timeout message at line 209, which reads as "the workers were slow" — it never says the dispatcher could not list at all. At the default budget that is 4 × 1200 s of blocking (see the runner-seam comment) for a pure permissions typo.
Cheapest fix that keeps the fail-open posture: remember whether the LIST has ever succeeded, and escalate to WARNING once when the first poll of a run fails, e.g.
except Exception as e:
logger.warning(f"stage fleet: cannot list {records_from} ({e}) — the barrier will time out")on the first occurrence, DEBUG thereafter; and include the "never listed successfully" fact in the timeout message so the two outcomes are not reported identically.
There was a problem hiding this comment.
🤖 from Claude
Fixed in 4e4ed40. _present now returns (basenames, listed_ok) and logs the failure at WARNING, not DEBUG; its docstring names the .status sibling's permission case as the likely one.
It also stops burning the budget: await_records counts CONSECUTIVE failures and abandons the barrier after _LIST_FAULT_LIMIT = 5 with a loud warning saying the dispatcher cannot SEE the records (permissions, region or endpoint), that waiting out the budget would buy nothing, and how many records it did see — returning fail-open, so the posture is unchanged. The timeout message now also appends "and the status prefix never listed successfully" when no poll ever succeeded, so the two outcomes no longer read identically.
Tests: TestBarrier::test_a_failed_list_is_not_an_empty_prefix, ::test_a_broken_list_ends_the_barrier_instead_of_burning_it (asserts the poll count equals the fault limit, i.e. it does not poll to the deadline), ::test_the_timeout_says_when_it_never_listed_at_all.
| "role": "finisher", | ||
| "run_id": run_id, | ||
| "records_from": records_from, | ||
| "touch_policy": touch_policy, |
There was a problem hiding this comment.
🤖 from Claude (review)
lease_ttl_s is threaded into every stage block but silently dropped from the finisher's.
{
"role": "finisher",
"run_id": run_id,
"records_from": records_from,
"touch_policy": touch_policy,
},Lines 308-309 do forward it for stage invokes:
if lease_ttl_s is not None:
block["lease_ttl_s"] = int(lease_ttl_s)and the handler reads it for both roles (deployment/aws/lambda_handler.py:1361 for the finisher, :1377 for a stage), passing it to run_stage_finisher, whose first act is acquire_lease(..., ttl_s=int(lease_ttl_s or DEFAULT_TTL_S)) (src/zagg/sweep_stages.py:1005-1011). So a run started with a non-default TTL re-admits its finisher on DEFAULT_TTL_S instead — the intent's TTL changes on the last acquire before release, which is precisely the window the finisher docstring calls out as "exactly the window a foreign sweep may claim" (sweep_stages.py:977-979). run_stage_sweep has no equivalent split: one ttl_s governs the whole run.
One line — add the same if lease_ttl_s is not None key to the finisher block.
There was a problem hiding this comment.
🤖 from Claude
Fixed in 2f56761. The finisher block is now built as a named dict and takes the same if lease_ttl_s is not None key as the stage blocks, with a comment on why (the finisher re-acquires before it releases, so dropping the TTL here moves it on the LAST acquire — widening exactly the window sweep_stages.py:977-979 calls out).
New test test_the_lease_ttl_reaches_the_finisher_too runs the fleet with lease_ttl_s=137 and asserts {b["lease_ttl_s"] for b in client.blocks()} == {137} across all four invokes, with the finisher last.
| """ | ||
| deadline = time.monotonic() + float(timeout_s) | ||
| while True: | ||
| seen = _present(records_from, store_kwargs) & expected |
There was a problem hiding this comment.
🤖 from Claude (review)
The barrier has no freshness test, so a supplied run_id that has been used before makes every barrier pass instantly on stale records.
seen = _present(records_from, store_kwargs) & expected
if seen >= expected:
return seen, False_present only asks "does an object with this basename exist under the prefix". The record names are deterministic from (dispatch, batch) — that is the design (stage_record_name, sweep_stages.py:674-682) — and the prefix is keyed on run_id alone (run_status_prefix). run_id is a documented parameter of run_stage_sweep_fleet ("generated here (or supplied)", line 240), and reuse is a natural thing to do: the lease is same-run-permissive by design, so re-driving a run that died mid-fan-out under its original id is the obvious recovery move.
Do that and every barrier returns timed_out=False on the previous attempt's records, immediately, before the new invokes have run. The tuples then fan out with no ordering at all — the fleet stops mirroring run_stage_sweep's ordering, which line 194-196 says is what the byte-identity acceptance rests on — and the finisher fires at once and aggregates a mix of old and new records.
Cheapest guard: snapshot _present(records_from, ...) once before the first invoke of a tuple and subtract it from seen, so the barrier waits on records that appear after the fan-out. Alternatively refuse a supplied run_id whose status prefix is already non-empty, and say so by name.
There was a problem hiding this comment.
🤖 from Claude
Fixed in bfd9b08, taking your cheaper guard rather than refusing a reused id — re-driving a dead run under its original id stays the supported recovery.
run_stage_sweep_fleet snapshots _present(records_from, ...) ONCE before the first invoke and threads it into every barrier as ignore=; await_records subtracts it from each poll, so a barrier waits on records that appear AFTER the fan-out. A non-empty snapshot logs an info naming the count. The finisher's barrier gets the same treatment, so a prior attempt's finisher.json cannot satisfy it either.
New test test_a_reused_run_id_cannot_pass_a_barrier_on_stale_records pre-seeds every record name the run will produce (all three stage records plus the finisher's) under an explicit run_id="reused", then drives the fleet with a client that lands nothing, and asserts every barrier expires. Verified it fails when ignore= is dropped.
| "run_id": "stage-20260825T094152Z-53c774", | ||
| "run_started": "2026-08-25T09:41:52+00:00", | ||
| "dispatch": 6, | ||
| "nodes": ["111", "112"], |
There was a problem hiding this comment.
🤖 from Claude (review)
The canonical JSON example is an event the worker refuses by name. Three of its fields cannot co-exist:
"leaves": [[1152921504606846982, null]]decodes to D1 decimal'1111111'—_decimal_order6, i.e. a shard_order-6 store.stage_tuples(6, tuple_width=3)(src/zagg/sweep_stage.py:173,for dispatch in range(0, shard_order, tuple_width)) yieldsdispatch3 and 0 only.dispatch: 6is not a dispatch order for that store at that width:>>> stage_tuples(6, tuple_width=3) [{'dispatch': 3, 'orders': [5,4,3], 'child_order': 6}, {'dispatch': 0, 'orders': [2,1,0], 'child_order': 3}]"nodes": ["111", "112"]are order-2 nodes (_decimal_baseis 1 char for an unsigned decimal,hive.py:933;_decimal_order('111') == 2). Againstdispatch: 6run_stage_workerraises before reading anything —src/zagg/sweep_stages.py:877:stage invoke (...) was handed nodes that are not at order 6: ['111', '112'] (orders [2]) — containment resolves both ways, so an off-order node over-claims its ancestor's whole subtree
The table one screen down states that rule correctly ("every entry must sit at exactly dispatch order — the worker refuses otherwise"), which makes the example the thing that is wrong, not the prose. An operator hand-building a probe event from this block gets a 500 with no obvious cause.
Self-consistent for the leaf shown: "dispatch": 3, "nodes": ["1111", "1112"] (_node_at('1111111', 3) == '1111'). If dispatch: 6 is wanted instead, the example needs an order-9 leaf and 7-char nodes.
There was a problem hiding this comment.
🤖 from Claude
Folded in e4da433. Confirmed your arithmetic before changing it:
leaf decimal: 1111111 order: 6
tuples(6,3): [(3, [5, 4, 3]), (0, [2, 1, 0])]
_decimal_order('111'): 2
_node_at('1111111', 3): 1111 order 3
The example is now "dispatch": 3, "nodes": ["1111", "1112"] — self-consistent for the order-6 leaf it ships with. Also extended the nodes table row to state the derivation for this example, so the reader can check the next one themselves rather than rediscovering it from a 500.
Generated by Claude Code
|
|
||
| ### Running it | ||
|
|
||
| Opt in on a Lambda-backed run with `output.sweep: "stages"` — the same knob |
There was a problem hiding this comment.
🤖 from Claude (review)
"Opt in on a Lambda-backed run" is true of exactly one of the three Lambda tails. _invoke_lambda_stage_sweep has one call site in the tree:
src/zagg/runner.py:4292— the spatial_run_lambdatail. Chained.src/zagg/runner.py:1632—RasterStrategy._run_lambda_shards(data_source.reader: raster, the S2 path). Calls_invoke_lambda_sweepand stops. Not chained.src/zagg/client.py:1236— the v2Run.dispatchtail. Callsrunner._invoke_lambda_sweepand stops. Not chained.
(src/zagg/runner.py:1194, the raster local tail, is likewise unchained, so "the same knob the local dispatcher reads" is also only the spatial local dispatcher at runner.py:3394.)
output.sweep: "stages" validates fine on any hive config (config.py:1303, must be a boolean or 'stages'), and get_sweep returns bool("stages") == True, so a raster or Run.dispatch run accepts the knob, runs the families sweep, and silently does no staged sweep. There is no warning at either seam. That is the failure mode a runbook exists to prevent — the operator's evidence is an absent ladder, hours later.
Either say which tail is wired ("the spatial Lambda tail"), or note the raster / Run.dispatch gap explicitly. Given the standing "raster full hive parity" posture I'd expect the gap itself to be an issue rather than a documented carve-out, but that is @-free scope for the author to raise, not something the doc can paper over.
There was a problem hiding this comment.
🤖 from Claude
Folded in 3c00028. Verified the call sites — _invoke_lambda_stage_sweep has exactly one (runner.py:4293, the spatial _run_lambda tail), while runner.py:1638 (raster) and client.py:1244 (Run.dispatch) call _invoke_lambda_sweep and stop.
The section now names the wired tail explicitly ("the same knob the spatial local dispatcher reads") and carries a warning admonition for the gap: the knob validates on any hive config and every tail runs the families sweep, but only the spatial Lambda tail dispatches the staged one — no warning at either seam, so the ladder silently is not built and the evidence is an absent ladder hours later. It points those paths at an explicit run_stage_sweep_fleet call or a python -m zagg.sweep --stages pass.
Agreed the carve-out is the wrong end state given the raster-parity posture, and agreed it is not the doc's to fix — leaving the "should this be an issue" call for review rather than opening one.
Generated by Claude Code
There was a problem hiding this comment.
🤖 from Claude
Follow-up to the fold above (bd469b3): the warning now separates the two unwired tails, because they are not the same case.
raster is unwired by construction, not by omission. A raster store is column-less — no digest columns to fold above the shard — so it declares no zagg-pyramid/2 block at all, and a stage worker handed one refuses at the declaration gate rather than building anything:
if not isinstance(pyramid, dict) or pyramid.get("spec") != PYRAMID_SPEC_V2:
raise ValueError(f"staged sweep requires a {PYRAMID_SPEC_V2!r} manifest pyramid declaration ...")(zagg.sweep_stage.ladder_entries, called by both run_stage_worker and run_stage_finisher.) Nothing is absent from such a store, so the raster-parity posture is not in tension here.
Run.dispatch is the real gap and the doc now says only that: a column-bearing /2 store dispatched through it accepts the knob, runs the families sweep, and silently builds no ladder. Whether to wire that tail is a scope call rather than something this PR should decide, so it stays standing as written — the runbook tells the operator to follow such a run with run_stage_sweep_fleet or python -m zagg.sweep --stages in the meantime.
| The one **documented** difference between executors is pinned by its own test: | ||
| the dispatcher derives dispatch nodes from the work set it holds, so a subtree | ||
| that appears only in the root MOC is not invoked. Both a post-run chained sweep | ||
| and a discovery-driven one hand over a work set that covers the MOC, so the |
There was a problem hiding this comment.
🤖 from Claude (review)
Both halves of this escape clause are wrong, and the first half contradicts the admonition 40 lines up.
Both a post-run chained sweep and a discovery-driven one hand over a work set that covers the MOC, so the identity claim holds wherever the transport is actually used.
(1) A post-run chained sweep does not hand over a work set that covers the MOC. Its leaves are leaves_from_stats_records(...) — this run's touched shards only (src/zagg/runner.py:4267). On an appended store the root MOC carries every prior run's shards too, so the work set is strictly narrower. That is precisely the case test_a_moc_only_subtree_is_the_one_documented_difference pins, and it is what the "Dispatch nodes come from the work set" note at line 597 already says. The two paragraphs cannot both be right.
The identity claim does hold for the chained case — but for a different reason. The local twin is scoped identically: stage_sweep_after_run passes scope = sorted({morton_decimal(int(k)) for k, _w in ...}), the run's own shard decimals (src/zagg/sweep_stages.py:641), so the in-process pass's work set ∪ MOC is filtered right back down to the run's footprint. Same node set, same bytes. The correct sentence is "both executors scope the chained sweep to the run's footprint", not "the work set covers the MOC".
(2) There is no discovery-driven fleet path. run_stage_sweep_fleet's only caller in the tree is runner._invoke_lambda_stage_sweep (grep for sweep_fleet outside the module itself returns only docstring references plus that seam), and it takes leaves as a required positional. The discover: true fallback in build_stage_event is not it either: that is per-batch worker-side re-derivation, and the worker still prefix-filters to the nodes the dispatcher already chose from its work set, so it cannot widen the node set. python -m zagg.sweep --stages is discovery-driven but in-process, not fleet.
As written this paragraph tells an operator the documented difference never bites in practice. It bites on every incremental append.
There was a problem hiding this comment.
🤖 from Claude
Folded in 0c69d06 — you are right on both halves, and the paragraph is rewritten rather than patched.
Verified your correction at the source: stage_sweep_after_run builds scope = sorted({morton_decimal(int(k)) for k, _w in ...}) (sweep_stages.py:641), the run's own shard decimals, so the in-process pass's work set ∪ MOC is filtered right back to the run's footprint — same node set, same bytes. And run_stage_sweep_fleet's only caller is the runner seam, with leaves a required positional; discover: true is per-batch worker-side re-derivation that still prefix-filters to the dispatcher's chosen nodes, so it cannot widen anything.
The text now (1) says the difference is real and bites on every incremental append, because the MOC carries every prior run's shards while the work set carries only this one's, and (2) attributes the chained case's identity to the scoping, not to coverage — with the closing note that driving the transport over a work set narrower than the store's coverage gives exactly what the test pins: the un-invoked subtree keeps its prior ladder and heals on the next pass that includes it. The contradiction with the line-597 admonition is gone.
Generated by Claude Code
| **Permissions.** Nothing new on the worker side — the execution role already | ||
| writes the store and the `<store>.status/` sibling (the issue #151 async result | ||
| channel), which is where the stage records land. The dispatcher needs | ||
| `lambda:InvokeFunction` plus `s3:ListBucket` scoped to `<store>.status/*` so it |
There was a problem hiding this comment.
🤖 from Claude (review)
The permissions paragraph is right about the conclusion and wrong about the grant. Two problems, one of which an operator will copy straight into a policy.
(a) s3:ListBucket cannot be "scoped to <store>.status/*". It is a bucket-level action: its Resource is the bucket ARN and the key prefix rides an s3:prefix condition. An operator following this sentence literally writes
- Effect: Allow
Action: s3:ListBucket
Resource: arn:aws:s3:::bucket/prefix.zarr.status/* # never matchesand gets a policy that matches nothing. What the barrier actually needs:
- Effect: Allow
Action: s3:ListBucket
Resource: arn:aws:s3:::bucket
Condition: { StringLike: { s3:prefix: "prefix.zarr.status/*" } }The failure is silent-ish and slow: _present warns per poll and await_records ends the barrier at _LIST_FAULT_LIMIT (5) consecutive faults — which reads, in the run log, exactly like the "cannot list the status prefix at all" case sweep_fleet.py:266-278 predicts.
(b) The dispatcher also GETs, not only LISTs. _read_finisher_record does obstore.get(store, FINISHER_RECORD_NAME) (src/zagg/sweep_fleet.py, called right after the finisher barrier). It is wrapped in a bare except that logs finisher record unreadable and returns {}, so a ListBucket-only dispatcher completes the run but silently loses stage_records / levels / lease / record / duration_s from summary["finisher"] — the only place the run reports what the finisher actually did. Add s3:GetObject on <store>.status/* to the enumeration.
What checks out: "the v2 Event transport already requires that prefix" is correct — src/zagg/client_transport.py:934 does obstore.list(self._store) and :961 does obstore.get(self._store, entry.key), both against the run status prefix — so the conclusion ("a correctly scoped dispatcher role needs no new grant") survives both fixes, including for GetObject. And "nothing new on the worker side" holds: <store>.status/ is a key-prefix sibling inside the same bucket, covered by ${OutputBucketName}/*, by the whole-bucket sliderule-public-cors/* grant, and by .../englacial/zagg/demo/* for a published store (deployment/aws/template.yaml:226-292). No template change is needed — that claim is accurate.
There was a problem hiding this comment.
🤖 from Claude
Folded in d20212e — this was the one an operator would have copied into a policy, so both halves are fixed.
(a) The paragraph now states that s3:ListBucket is a bucket-level action whose Resource is the bucket ARN with the prefix on an s3:prefix condition, and ships your wrong/right pair as a literal YAML block so the non-matching key-level ARN is shown as the trap it is.
(b) s3:GetObject is added to the enumeration, with _read_finisher_record's consequence named: the run completes and silently loses stage_records / levels / lease / record / duration_s from summary["finisher"].
The conclusion survives both, as you say, and I kept it: the v2 Event transport already lists and gets that prefix, so a correctly scoped dispatcher role needs no new grant and no template change ships. Also carried over your point that both failures are quiet — a missing list grant reads as slow workers until the five-fault cutoff, a missing get grant just drops the reporting.
Generated by Claude Code
| | `dispatch` | stage | The tuple's dispatch order. The worker runs exactly that one tuple | | ||
| | `nodes` | stage | This invoke's dispatch nodes, as morton decimals. Must be non-empty and every entry must sit at exactly `dispatch` order — the worker refuses otherwise | | ||
| | `batch` | stage | Which batch of that tuple this is; names the record object | | ||
| | `tuple_width` | both | Optional; defaults to `zagg.sweep_stage.DEFAULT_TUPLE_WIDTH` | |
There was a problem hiding this comment.
🤖 from Claude (review)
Two Role-column errors in the table.
tuple_width is stage, not both. run_stage_finisher has no tuple_width parameter (src/zagg/sweep_stages.py:941-951), and _handle_stage_sweep does not pass one on the finisher branch (deployment/aws/lambda_handler.py:1356-1364 passes only run_id, records_from, touch_policy, lease_ttl_s, store_kwargs). A tuple_width on a finisher block is inert. zagg.sweep_stage.DEFAULT_TUPLE_WIDTH as the default is correct (= 3, and the handler does int(block.get("tuple_width", DEFAULT_TUPLE_WIDTH)) from zagg.sweep_stage, singular — the module path in the table is right).
records_from is required for the finisher and optional for a stage worker — the row reads as uniformly optional-ish, and both ends of that asymmetry are load-bearing:
- finisher:
if not records_from: raise ValueError(...)(sweep_stages.py:999) — refuses by name, as does a prefix that lists zero records. - stage:
if records_from:guards the PUT (sweep_stages.py:935). A stage invoke without it folds correctly and writes no record at all, so its barrier waits out the fullbarrier_timeout_sand the finisher's per-level actuals under-report — a hand-built event that omits the key looks like a lost invoke.
Every other optional key in the table says "Optional" in the Description; these two deserve the same explicitness. Same for batch (defaults to 0) and touch_policy (defaults to "auto"), neither of which is marked.
There was a problem hiding this comment.
🤖 from Claude
Folded in 8994119. Both confirmed against the source before editing — run_stage_finisher (sweep_stages.py:941-951) has no tuple_width, and the handler's finisher branch passes only run_id / records_from / touch_policy / lease_ttl_s / store_kwargs.
Table now says: tuple_width role stage, with the note that one on a finisher block is inert; records_from required for the finisher (refuses by name, as does a zero-record prefix) and optional for a stage worker, with the consequence spelled out — folds correctly, writes no record, so its barrier waits out the full timeout and looks exactly like a lost invoke; and batch (defaults 0) and touch_policy (defaults "auto") both marked optional.
Generated by Claude Code
There was a problem hiding this comment.
🤖 from Claude
Follow-up to the fold above: the code half landed too, in c092583.
records_from is now required for both roles, not just the finisher. run_stage_worker refuses by name before admission and before any fold:
stage invoke for run 'F' (dispatch 0, batch 0) was handed no records_from — a worker
whose record nobody can witness reads to the dispatcher exactly like a lost invoke,
and its coverage never reaches the manifest; refusing rather than folding invisibly
The reasoning is your own: an invoke that folds correctly and writes no record is strictly worse than a loud refusal — the dispatcher cannot tell it from a lost one, it burns the full barrier timeout, and its coverage silently never reaches the manifest. The dispatcher always sends the key (zagg.sweep_fleet.run_stage_sweep_fleet puts it on every block, stage and finisher alike), so nothing that exists today loses a capability.
Test: test_a_worker_nobody_can_witness_refuses_by_name — asserts the refusal AND that it happens early enough to leave no lease and no artifacts. The table row now reads "Required" for both roles, with the finisher's extra zero-records refusal kept as the separate condition it is.
| (default 7,200 s), so the tail's worst case is a constant rather than a | ||
| function of the store's order. Past the total, each remaining barrier degrades | ||
| to a single check. An expiry is recorded as `barrier_timed_out` on the run | ||
| summary and in the finisher's stage block: when the invoke was merely *queued* |
There was a problem hiding this comment.
🤖 from Claude (review)
barrier_timed_out reaches the finisher's wire block and is then dropped. The dispatcher sets it:
finisher_block = {
"role": "finisher", "run_id": run_id, "records_from": records_from,
"touch_policy": touch_policy,
# ... "it rides on the wire so the RUN record says so."
"barrier_timed_out": stages_timed_out,
}(src/zagg/sweep_fleet.py, run_stage_sweep_fleet)
But _handle_stage_sweep never reads it (deployment/aws/lambda_handler.py:1356-1364) and run_stage_finisher has no such parameter (src/zagg/sweep_stages.py:941-951). Grepping barrier_timed_out across src/, deployment/ and tests/ shows it only in sweep_fleet.py, runner.py:5660 (the dispatcher's own log line), and dispatcher-summary assertions in test_sweep_stage_fleet.py — never in a stage/finisher record. So it is recorded on the run summary (true, and only in the dispatcher's process) but not in anything durable; the run record does not say its actuals may be short.
Two things follow for this section:
- The sentence "and in the finisher's stage block" overstates what survives the invoke. Either scope it to the dispatcher summary, or say the key rides the wire but the handler currently ignores it.
- The wire-grammar table above omits
barrier_timed_outentirely, so it is not the complete grammar it presents itself as. A key the dispatcher always sends on a finisher event belongs in the table — with whatever its real status turns out to be.
(The underlying gap is phase 3's, not this doc's — the fix may well be to thread it into run_stage_finisher rather than to soften the sentence.)
There was a problem hiding this comment.
🤖 from Claude
Folded the doc half in bf61344; the code half I am deliberately leaving standing.
Confirmed the gap: run_stage_finisher (sweep_stages.py:941-951) has no barrier_timed_out parameter and _handle_stage_sweep never reads the key, so the dispatcher's finisher_block["barrier_timed_out"] is inert on the wire.
Both of your doc points are now addressed: the prose says the expiry is recorded on the run summary and only there — the dispatcher does put the key on the block, the handler does not forward it, nothing durable records that the actuals may be short — and directs the operator to the dispatcher log. The key is also in the wire-grammar table now, marked "currently inert", so the table is complete.
What I have not done is your parenthetical — threading it into run_stage_finisher so the run record carries it. That is a behaviour change to phase 1's finisher signature and record grammar, not a diff-scoped doc fix, and there are two defensible answers (thread it through and record it, vs. drop the key from the dispatcher's block as dead weight). Leaving the choice for review rather than picking one; the doc is accurate either way, and if it gets threaded through the table row is the one line to update.
Generated by Claude Code
There was a problem hiding this comment.
🤖 from Claude
Reversing the leave-standing above: the code half landed in 4584761.
Threading it through was the right of the two answers, for the reason your finding gives — dropping the key would leave the only witness in a dispatcher log that dies with the process, and the manifest would then claim coverage nobody observed. So:
run_stage_finishertakesbarrier_timed_out: bool = Falseand records it on the summary, hence in the store-root run record (sweep_stats_{ts}_stages.json) and infinisher.json;- the handler forwards
block.get("barrier_timed_out"); - a true value also logs a WARNING naming the run and the record count it aggregated from.
It is recorded present and false on a clean run, never absent, so a reader never has to guess whether an old finisher simply did not write it.
Tests: test_an_expired_barrier_is_recorded_in_the_run_record, test_a_clean_run_records_the_barrier_as_intact, and test_the_dispatchers_barrier_verdict_reaches_the_finisher (end to end over the wire — dispatcher block to handler to durable record).
The table row is no longer "currently inert", and the sequencing paragraph now says the expiry is stamped durably rather than pointing the operator at a log.
| `tests/test_sweep_stage_fleet.py` builds a column-bearing store, runs the CLI | ||
| staged sweep, snapshots every object, resets the store to its pre-sweep bytes, | ||
| runs the fleet path with the Lambda client mocked to execute the worker arm | ||
| in-process, and byte-compares every object — chunk data exactly, JSON modulo |
There was a problem hiding this comment.
🤖 from Claude (review)
"byte-compares every object" is slightly wider than the oracle. _artifacts drops run records before comparing:
def _is_run_record(rel: str) -> bool:
return (rel.startswith("sweep_stats_") and rel.endswith("_stages.json")) or (
rel.startswith("stats_") and rel.endswith(".parquet")
)(tests/test_sweep_stage_fleet.py:1490, filtered out in _artifacts)
They are not unchecked — test_both_arms_leave_one_run_record_and_no_lease asserts one of each and no lease — but they are not byte-compared, and "every object" reads as if they were. Separately, the cross-width arm (test_identity_survives_a_different_tuple_width) compares with ladder_data_only=True: array chunks and array metadata only, dropping each artifact's group zarr.json, because source_children and the summed generation are legitimately width-dependent. The parametrized (1, 3, 181) / (2, 2, 163) / (3, 1, 136) arms are full comparisons, so the "tuple widths 1, 2 and 3 (136–181 objects each)" claim on the next paragraph is exact — it is only the cross-width arm that is narrowed.
Suggest "byte-compares every store artifact — chunk data exactly, JSON modulo the run identity and the clock; the run records are checked for shape rather than bytes."
Verified accurate in this paragraph and the next, for the record: the object counts (@pytest.mark.parametrize("width,tuples,objects", ((1, 3, 181), (2, 2, 163), (3, 1, 136))), line 1716); the snapshot → _cli_sweep → _restore → fleet → snapshot shape; _canonical's split (JSON/.moc devolatilized, everything else raw bytes); all four listed arms (test_identity_survives_a_multi_batch_fan_out, _DeferringLambda, test_identity_holds_for_a_windowed_store, test_identity_holds_for_a_both_channel_store); and both negative controls (test_the_oracle_detects_a_wrong_fold, test_the_oracle_detects_a_divergent_build).
There was a problem hiding this comment.
🤖 from Claude
Folded in f3423db, taking your suggested wording. "byte-compares every object" is now "byte-compares every store artifact — chunk data exactly, JSON modulo the run identity and the clock", with the run records called out as checked for shape rather than bytes (one of each, no leftover lease) because they differ between executors by design.
Also closed the second half: the parametrized width arms are now described as "both executors at the same width so the comparison stays full", and the cross-width arm gets its own sentence naming it as the deliberately narrower one — group zarr.json dropped because source_children names a genuinely different fold across widths, with those attrs re-checked separately so nothing else hides behind the exclusion.
Thanks for the verification pass on the rest of the paragraph; the object counts and the arm list are unchanged.
Generated by Claude Code
| *scheduling* preference, not a correctness device: under-coverage is | ||
| recorded in each artifact's own `source_children` and heals on the next | ||
| pass, so an expired barrier logs loudly and the run proceeds; | ||
| 3. next tuple; then the **finisher** invoke last — root `coverage.moc`, |
There was a problem hiding this comment.
🤖 from Claude (review)
The zero-dispatch-node early return is missing from the sequencing, and it is operator-visible. run_stage_sweep_fleet short-circuits before the finisher when no tuple produced a dispatch node:
if not summary["stages"]:
# ... Firing it would buy one guaranteed 500 the Event invoke hides, then a
# full barrier waiting on a record that can never land.
summary["skipped"] = "no dispatch nodes"
summary["finisher"] = {"landed": False, "fired": False}
return summary(src/zagg/sweep_fleet.py, after the tuple loop)
As written, step (3) says the finisher fires last, unconditionally. A run whose leaves were all skipped by _normalize_leaves (mixed-order) or filtered out by scope returns a clean-looking summary with zero invokes and finisher.fired: false — which is a correct outcome, but the operator reading this runbook has no way to tell it apart from "the finisher was lost". One clause on step (3) ("…then the finisher invoke last, unless no tuple produced a dispatch node, in which case nothing fires at all and the summary says skipped: "no dispatch nodes"") closes it.
There was a problem hiding this comment.
🤖 from Claude
Folded in 32c3179. Step (3) now carries the clause: the finisher fires last unless no tuple produced a dispatch node — every leaf skipped as mixed-order, or filtered out by scope — in which case nothing fires, there is no finisher and no barrier, and the summary says skipped: "no dispatch nodes" with finisher.fired: false. Kept the reason too (a finisher over zero records refuses by design, so firing it buys one guaranteed 500 the Event invoke hides plus a barrier on a record that can never land), since that is what tells the operator it is a correct outcome rather than a lost invoke.
Generated by Claude Code
| pool for all users of the function, so it is off by default and independent | ||
| of the self-recycle knobs (both can be enabled). | ||
|
|
||
| ## Staged pyramid sweep over the fleet {#staged-sweep} |
There was a problem hiding this comment.
🤖 from Claude (review)
Nit: the new {#staged-sweep} anchor has no inbound link. The narrative home of the staged sweep is docs/hive_layout.md:512 (### The staged sweep (issue #384)), which three other places in that file already point at (:338, :457, :508) — and which says the sweep is "chained immediately after a fleet run with the opt-in output.sweep: "stages"" without ever mentioning that a fleet transport for it now exists. A reader who arrives via the /2 ladder narrative never learns this page is here.
Every other anchor added to lambda.md is linked from somewhere (#write-probe from hive_layout.md:236). One line in hive_layout.md's staged-sweep section pointing at deployment/lambda.md#staged-sweep would match that convention. Not a correctness finding — flagging it because it is cheap and this section is otherwise unreachable except by scrolling.
There was a problem hiding this comment.
🤖 from Claude
Folded in d61513d. Added the inbound link at the narrative home — docs/hive_layout.md's staged-sweep section now ends with a paragraph naming the worker-side transport and pointing at deployment/lambda.md#staged-sweep, matching the #write-probe convention you cite.
Generated by Claude Code
|
🤖 from Claude All four phases are landed and every review finding is folded. Applying What landed this runPhase 4 (docs + deployment notes) plus the fold of the phase-4 self-review's nine findings, all on
Every one of those has a reply on its thread. Three were substantive enough to call out: the canonical JSON example was an event Gate
Two notes on processThis PR was worked concurrently by two runs. Phases 2 and 3 and the twelve phase-3 findings landed from the other run while this one was folding the same findings locally; I discarded my duplicates rather than force-pushing over anything, and rebuilt from Left in draft deliberately. §2/§5 say a PR whose phases are all complete should be flipped ready-for-review when What needs a decision
Nothing else is outstanding: the checklist is complete, all twenty-one review findings across phases 1–4 are folded, and Generated by Claude Code |
|
🤖 from Claude Gate confirmation, closing out the run — both the local suite and CI finished after my status note above, so the numbers there are now settled rather than provisional. Local, full suite on the pushed head (d61513d): CI: No change to the state: still draft, still Generated by Claude Code |
2 participants
Closes #519.
The
/2staged dense sweep (issues #384/#416) is complete and tested but ran only in the local CLI process. On a source.coop-published store there is no sanctioned local write path at all — the bucket policy names the fleet execution role as the write identity (#495/#496) — so above-shard overviews there were stuck on the/1cascade retrofit. This PR adds the D8 worker transport: the staged sweep fans out over Lambda, the dispatcher invokes and polls and never writes, and the fleet-built ladder is byte-identical to the CLI-built one.Implementation follows the plan comment on the issue: #519 (comment)
Approach
One event, one extra block. No new mode:
mode: "sweep"grows an optionalstageblock, so the credential resolution (_output_store_kwargs), the work-set transport (leavesinline /discover: true) and the async-payload budget are the ones the rollup-families arm already uses.The wire grammar as landed
{ "mode": "sweep", "store_path": "s3://bucket/prefix.zarr", "leaves": [[1152921504606846982, null]], // or "discover": true "output_credentials": { … }, // optional, unchanged "stage": { "role": "stage" | "finisher", // default "stage" "run_id": "stage-20260825T094152Z-53c774", "run_started": "2026-08-25T09:41:52+00:00", // stage; dispatcher-pinned "dispatch": 3, // stage; the tuple's dispatch order "nodes": ["1111", "1112"], // stage; morton decimals, all at `dispatch` "batch": 0, // stage; names the record object "tuple_width": 3, // optional; DEFAULT_TUPLE_WIDTH "partition": {"index": 0, "of": 4}, // optional, recorded only "lease_ttl_s": 900, // optional "records_from": "s3://bucket/prefix.zarr.status/run-stage-20260825T094152Z-53c774", "touch_policy": "auto", // finisher (issue #501) "barrier_timed_out": false // finisher; the dispatcher's verdict } }records_fromis required for both roles, and both refuse by name without it.nodesmust be non-empty and every entry must sit at exactly thedispatchorder —scope_admitsresolves containment in both directions, so a coarser or finer node silently widens the invoke's share, and two same-run siblings would then double-write (neither the lease nor the foreign-fresh stamp catches that: both are same-run-permissive by design, which is how a fan-out's siblings are admitted). A bad node set has to be loud there or it is never loud at all.The worker arm (
zagg.sweep_stages.run_stage_worker) runssweep_stage_passrestricted to one tuple (only_dispatch=) and to the nodes it was handed (they reach the pass as the ordinaryscopeMOC). Everything else is the in-process pass, unchanged: lease admission, the run-id skip keys, theForeignSweepErrorbackstop, every store write worker-side. Splitting a tuple across invokes is free because dispatch nodes at one order own disjoint subtrees and a tuple's folds read only columns one tuple finer.The lease is the existing per-store admission lease. The first stage worker creates the intent; siblings of the same run read their own back (idempotent re-admission); a live foreign intent refuses the invoke by name. The finisher takes the lease before it writes and releases it as its final act, so a run that dies mid-fan-out leaves a claimable intent, not an open store.
The stage record is the fleet's aggregation channel and its soft-barrier signal. Each invoke PUTs
stage-<dispatch>-<batch>.jsonunder the run's status prefix (<store>.status/run-<run_id>/, a store sibling), carrying its per-stage rows and its raw per-artifactlevel_actuals. The dispatcher names every record before firing, so the barrier needs no listing of unknowns and no worker response. The finisher reads them back — filtered to its ownrun_id— to rebuild the actuals the in-process driver keeps in memory. Merging is exact:_accumulate_actualskeys rows per(artifact node, window)and assigns, never adds, so a coarse ancestor two batches both visited contributes its row once.The dispatcher (
zagg.sweep_fleet.run_stage_sweep_fleet, new module) mirrorsrun_stage_sweep's tuple ordering: fan out a tuple's nodes batched under the 250 KB cap → soft-barrier on the stage records → next tuple → finisher last. One barrier is bounded bybarrier_timeout_s(2,700 s = 3× the function timeout), the sum bytotal_barrier_budget_s(7,200 s), so the tail's worst case is a constant rather than a function of the store's order. Dispatch nodes come from the work set the dispatcher holds, never from the store — an invoke-only role cannot read the root MOC, and D8 keeps it that way.The runner seam is
_invoke_lambda_stage_sweep, called from the spatial_run_lambdatail underoutput.sweep: "stages", fail-open (D9).The byte-identity oracle (the acceptance)
TestByteIdentityOracleintests/test_sweep_stage_fleet.py, fully offline — the Lambda client is mocked to execute the worker arm in-process. One store, swept twice from the same pre-sweep bytes: build a column-bearing store → run the real CLI (zagg.sweep.main([root, "--stages"])) → snapshot every object → restore the store byte-for-byte → run the fleet path → compare. Chunk data exactly; JSON modulo the run identity (run_id,run_ids) and the clock (written_at,generated_at,timestamp).Result: byte-identical at every tuple width.
source_childrenonlyThree negative controls prove it is not vacuous: a wrong fold (
merge_level_actualsdropping a child row) fails the byte branch, a scoped-out subtree fails the object-set branch, and stubbing the barrier out fails the deferring arm. Reviewers also confirmed by mutation that reversing the tuple order fails the width-1 and width-2 byte-exact arms, and truncating a batch's leaf refs fails the multi-batch arm.One documented, tested executor difference: the dispatcher derives dispatch nodes from the work set, so a subtree present only in the root
coverage.mocis not invoked (test_a_moc_only_subtree_is_the_one_documented_difference). That is the same scoped posturestage_sweep_after_runalready has (#381 point (11)); a worker still folds every child on disk under each node it is handed.Phases
64952f9e+ 9 folds). Thestageblock,run_stage_worker/run_stage_finisher,only_dispatch=onsweep_stage_pass,merge_level_actuals/read_stage_records, handler routing, wire grammar.3c79d0d0+ 12 folds).zagg.sweep_fleet, the batching, the bounded soft barrier, the runner tail seam.0a69c432+ 12 folds). The table above.3dbcbf53+ 11 folds). A "Staged pyramid sweep over the fleet" runbook indocs/deployment/lambda.md(wire grammar, sequencing, permissions, the trust argument, the deferral), linked fromdocs/hive_layout.md.Review
Four adversarial-review passes (one per phase, fresh context, Opus) posted 39 inline findings; all 39 are folded, one commit each, with a reply on every thread. The substantive ones: the node-set validation gap that allowed a whole-store sweep from an empty
nodes(phase 1); the finisher running with no lease at all (phase 1); an O(n²)pack_batchesand a byte-accounting undercount that silently degraded batches todiscover: true(phase 2); barriers passing instantly on a prior attempt's records (phase 2); andrun_idsmissing from the oracle's volatile set, which was blocking full byte identity (phase 3 — taking that upgrade is what turned the cross-width-only comparison into the table above).Two threads were left standing by the phase-4 fold and both code halves have since landed (
c0925839,4584761d):records_fromrequired for the stage role, andbarrier_timed_outthreaded into the durable run record.Sibling merged forward. PR #522 landed on
mainmid-run withzagg.store.put_objectand a build-failing guard against rawobstore.put. Merged forward (7f18d889, regular merge) and routed the stage record through the ACL-carrying handle (1b8cd96d) — not just to satisfy the guard: stage records land under the run's status prefix, which on a published store sits inside the same grant as the store, so a raw write would have published an object Source Cooperative cannot manage.Testing
tests/test_sweep_stage_fleet.py(new): 85 tests. Full suite4964 passed, 44 skipped(post-merge withmain). CI green at1b8cd96d(ruff, build x86_64 + arm64, test 3.12, test 3.13).Two pre-existing failures on
main, not touched:tests/test_lambda_build.py::TestFunctionBuild::test_function_build_succeeds(persistent locally) andtests/test_client_transport.py::TestStatusPoller::test_invoke_fault_burns_an_attempt_and_retries(flaky — it failed one CI run on this branch and passed on a plain re-run of the same sha, and it fails intermittently locally onmaintoo).docs/specification.mdneeds no change: this touches no wire format, attrs grammar, orspecmarker — the transport writes the same artifacts through the same writers — and the stage records live under the.statussibling, which §5.2 already treats as outside the store's hash scope.No AWS was touched. No invokes, no deploys, no S3 writes — the fleet is mid-campaign and the acceptance is offline by design. Live fleet validation is deferred and recorded: it rides the next release plus a cents-scale probe on a column-bearing probe store (the SERC GEDI 0.50 probe store is exactly that testbed). Nothing under
deployment/aws/other than the handler's stage arm was modified — no template, layer, or IAM change ships with this.Questions for review
_run_lambdatail chains the staged sweep. The raster tail is correct to be unwired by construction — a raster store is column-less, declares no/2ladder, and a stage worker handed one refuses at the declaration gate (ladder_entries), so nothing is missing from such a store. The v2Run.dispatchtail insrc/zagg/client.pyis the real gap: a column-bearing/2store dispatched through it acceptsoutput.sweep: "stages", runs the families sweep, and silently builds no ladder, with no warning at the seam.client.pywas outside this PR's territory (column backfill for pre-column stores: the /1 -> /2 upgrade bridge (columnize pass + /2 retrofit declaration) #520/published-bucket store handle 403s on LIST: the #495 canned-ACL default header is sent unsigned on non-PUT requests (blocks all source.coop fleet builds/sweeps) #522 in flight). Wire it here, or file it?_put_stage_recordand the handler's stage arm are deliberately not fail-open, unlike the families arm. A swallowed failure in a fan-out with a soft barrier is indistinguishable from a lost invoke, and would silently under-report coverage in the one place pyramid v2 (umbrella): leaf-push columns, (node, cells) level grammar, staged dense sweep #381 point (7) exists to record it. The run stays fail-open at the dispatcher's call site. Confirm that trade.Nonefor fleet workers (store-granular by correctness; recording one batch's node list would record whichever worker won the create race), so a fleet run's lease records no scope where a CLI run records one. Acceptable, or should the dispatcher's run scope be threaded through the finisher for the record?run_stage_sweep_fleetcall handed a partial work set folds less than the CLI would. Both real callers (post-run chaining, discovery-driven) are fine. Worth a guard, or is the documented precondition enough?