Follow-ons to #183: dispatcher-side shard-aware granule_workers, and function-size variants gated on utilization

[espg]

🤖 from Claude

Follow-up to PR #183 (granule-level read concurrency, Closes #180 / #179), filed at espg's direction. Two items were deliberately left out of that PR; this issue captures them so they don't live only in the PR's "Questions for review". Both are gated on the fleet A/B at K∈{1,2,4,6} (issue #180 phase 3, espg-side): the per-invocation cpu_seconds telemetry that PR #183 adds is what produces the utilization data these decisions key on.

1. Dispatcher-side shard-aware K (per-shard granule_workers override)

PR #183 ships data_source.granule_workers as a config-level knob only: one K for the whole run. The spec (issue #180 phase 2) sketched the finer version:

Shard-aware K (free 80% of "shardmap-scoped workers"): the dispatcher can thread a per-shard granule_workers override into the cell event from the shardmap's granule counts — precedent: the aoi_payload per-cell extra in runner.py's _cell_work.

The worker side already accepts event-injected values (the plumbing seam exists in PR #183); nothing sets it yet. The point: shardmaps record granule counts per shard, so the dispatcher can clamp K to min(K_config, n_granules) — a 5-granule shard shouldn't spin up K=6 threads — and could go further (scale K down on the densest shards where RSS headroom at 2 GB is the binding constraint, per issue #117).

Scope sketch: runner.py _cell_work threads granule_workers into the cell event alongside aoi_payload; simplest policy first (min(K, n_granules)), anything cleverer only if the A/B shows shard heterogeneity matters. Tests: dispatcher unit test on the event payload; worker already validates event values.

Gate: worth doing at all only if the A/B lands on a default K > 1.

2. Function-size variants (different-size Lambdas) — separate, harder-gated decision

From issue #180's follow-on section, espg's criterion verbatim:

Lambda bills GB-seconds → bigger workers (vCPU ∝ memory, ~1/1769 MB, max 6) are cost-neutral whenever wall drops proportionally. Scale up only once K-in-flight work demonstrably saturates 2 vCPUs ("I don't want a 16-core lambda unless we hammer the CPUs" — espg). If shard heterogeneity warrants, function-size variants (2/4/7 GB) + dispatch routing by granule count.

Decision tree from the A/B's cpu_seconds/wall/RSS data:

1. Wall keeps dropping with K, utilization stays low (latency-bound reads) → raise K, keep 2 GB, close this item as not-needed.
2. RSS caps K before CPU does (memory-bound; K multiplies in-flight granule footprint ~20–40 MB decoded + digest state, and dense shards are already marginal at 2 GB per Order-10 single-shard processing OOMs at 2 GB (gain_bias) — 2 GB is marginal for order-10 #117) → one probe run at 3–4 GB: the memory bump buys K headroom and vCPU roughly for free if wall follows.
3. Utilization approaches the vCPU count (CPU-bound decode/aggregation) → the real design discussion: 2/4/7 GB function variants + dispatch routing by granule count (which composes with item 1 — the same shardmap granule counts drive both the K override and the size routing).

Gate: hard-gated on the A/B utilization data crossing the hammer-the-CPUs bar; do not start this on speculation. Touches deployment/aws/template.yaml (new function resources) — per repo convention that work needs an issue naming it, which this issue does, but the CloudFormation/deploy execution stays espg-side.

Refs

• Issue Granule-level read concurrency: the worker's serial granule x group loop is the dominant read-wall term #180 (spec; phases 1–3 landed in PR Granule-level read concurrency (K in flight, ordered fold) #183), issue Selection reads (segment-rate geolocation + link arrays) are serial, uncompiled h5coro on every backend #179 (folded into the same PR), issue Order-10 single-shard processing OOMs at 2 GB (gain_bias) — 2 GB is marginal for order-10 #117 (2 GB marginality on dense shards), issue add 88 south stress test shard + benchmark #148 (88S stress matrix — the natural A/B target alongside the NEON benchmarks).
• Related but distinct: the process-global chunk-fetch budget is upstream h5coro-hidefix work (spec parked on the PR thread: Granule-level read concurrency (K in flight, ordered fold) #183 (comment)) — if it lands, K×read_workers products stop being a connection-pool concern and only the RSS/CPU axes above remain.

(Left unlabeled for espg to triage — add plan/implement when the A/B data is in.)

[espg]

Gate: worth doing at all only if the A/B lands on a default K > 1.

default for k will be 4 (see table at #183 )

[espg]

scaling results are also positive (same link above) for larger workers

[espg]

what from this could / should be wrapped into the #185 PR?

[espg]

🤖 from Claude

What's wrappable into the #185 PR, item by item:

Already in it (the PR is being built now on claude/185-inline-selection-heuristic):

• Default granule_workers: 4 — your call on the K-sweep table, phase 2 of that PR.
• Item 1's simple clamp — min(K, n_granules) threaded into the cell event from the shardmap's granule counts in runner.py _cell_work, phase 3. So a 2-granule shard never spins 4 threads at the new default.

Could be wrapped, your pick — the MemorySize default bump. With the K=4 read (~53 s) closing on agg+finalize, one size step is the cheap test of your GB-seconds criterion, and it's a one-parameter template.yaml default change (this issue names the file, and deploying it stays yours — the PR only changes the default). One-pass question: (a) 3538 MB (exactly 2 vCPU — the cheapest 2× step; cost-neutral iff wall roughly halves at K=4–6), (b) 4096 MB (2.3 vCPU + RSS headroom for the dense #117 shards in the same move), or (c) keep it out and decide after a one-off 4 GB probe run on the current stack. Say (a)/(b)/(c) and the PR picks it up as a final phase — or not.

Should NOT be wrapped (stays gated here):

• Function-size variants + dispatch routing by granule count — new CloudFormation resources plus a dispatcher routing design; different risk class from a regression-fix PR, and the single-size bump above likely answers whether it's even needed (if one size fits after the bump, variants are dead weight).
• RSS-aware per-shard K (item 1's cleverer half) — the flat memory in your table (616 MB at K=4, +56 MB over K=1) says the streaming fold already killed the K×footprint fear; no evidence yet that any shard needs a lower K than the clamp gives.
• The process-global chunk-fetch budget — upstream h5coro-hidefix work (spec parked on the PR Granule-level read concurrency (K in flight, ordered fold) #183 thread); note K=4 saturating at the fetch budget is consistent with it working as designed, and a 2-vCPU worker would shift that ceiling, another thing the size-bump probe shows for free.

---

Generated by Claude Code

[espg]

🤖 from Claude

Resolving my own one-pass question above: the answer is (c) for now — #186 (sharded CSR write race, probabilistic in exactly the direction more vCPU makes worse) explicitly holds the memory bump until the race is fixed ("the memory bump must not ship first"). The #185 PR stays code-only (heuristic + K=4 default + clamp, plus #186's ask (1) observability rolled in per espg); the MemorySize default bump re-opens here once #186's root cause lands.

---

Generated by Claude Code

[espg]

follow on from #187 --

The K=4 default reaches direct process_shard callers (no dispatcher) unclamped — a 1-granule hand-rolled call now constructs a 4-wide pool (threads spawn lazily, so the cost is one idle pool object). Worth a worker-side min(K, len(granule_urls)) too, or keep the clamp dispatcher-only as agreed on #184?

...to be clear, we already check for this at the dispatcher?

[espg]

🤖 from Claude

Yes — the min(K, n_granules) clamp runs at the dispatcher, on both dispatch paths, before any worker is invoked:

• Lambda: _run_lambda calls _clamped_data_source(config.data_source, len(granule_urls)) (src/zagg/runner.py:1416); the clamped width rides in that cell's event config, otherwise the shared config passes through.
• Local: the same clamp is threaded into each cell's per-cell config (runner.py:1154-1157).
• It counts resolved urls (_resolve_urls — what the worker actually reads), not raw records, so href-less records can't inflate K (PR Inline selection-read heuristic + granule_workers default 4 #187 review finding).
• _clamped_data_source (runner.py:970) returns a clamped copy only when clamping lowers K, else None → unclamped cells keep the shared config byte-identical (covered by TestClampedDataSource and test_{local,lambda}_clamps_granule_workers_per_cell in tests/test_runner.py).

So every dispatched cell is clamped. The only unclamped path is a direct, hand-rolled process_shard(...) call that bypasses the dispatcher entirely (no shardmap → no per-cell granule count) — the residual case from the PR #187 thread. There a 1-granule call at the K=4 default constructs a 4-wide pool object, but threads spawn lazily, so the cost is one idle pool, not four idle threads (src/zagg/processing/worker.py:62-63 documents the dispatcher clamp).

That leaves the one open decision from PR #187 for your call: add a worker-side min(K, len(granule_urls)) as belt-and-suspenders too, or keep the clamp dispatcher-only as agreed here? Leaving waiting on your answer.

---

Generated by Claude Code