Dispatcher abstraction as shared infrastructure (#20, #12)

[espg]

The most impactful single investment. Both temporal (#12) and future cluster backends (#20) need the same generic fan-out → retry → measured-cost loop that currently lives as two bespoke functions in runner.py. A dispatch.py with a clean Executor protocol extracts this once. Every pipeline type (spatial, temporal, event) then inherits Lambda and local execution for free, and items 3–4 of #20 (vaex, ray/dask/slurm) plug in behind the same interface without touching pipeline logic.

This is the prerequisite that unblocks both #12 Phase 1c and #20 items 3–4.

[espg]

What's the main things we need to think about in designing this? What are the major decisions that need to be made?

[espg]

🤖 from Claude

Agreed this is the highest-leverage single investment — both #12 and #20 items 3–4 collapse into "another Executor implementation" once the loop is extracted. A few clarifying questions, then three design options for the Executor protocol's surface area.

Where the duplication actually lives today

In runner.py as of main (lines for grounding):

• _run_local(...) at ~307 — sequential per-shard loop, no retry/cost machinery (it's the trivial executor).
• _run_lambda(...) at ~395 — the rich one: ThreadPoolExecutor + boto3 invoke + retry + log parsing + cost estimation + concurrency probe (the post-Concurrency-aware Lambda orchestrator: pre-flight probe + FD-exhaustion guard #41 surface).
• _invoke_lambda_setup / _invoke_lambda_finalize / _invoke_lambda_cell at ~643 / ~671 / ~690 — the per-cell hooks.
• Plus concurrency.py (probe / FD ceiling / errno-24 guidance) which _run_lambda pre-flights.

So the "generic" loop has these moving parts: dispatch (sync or fan-out), retry policy, per-cell error classification (errno-24 special-cased today), per-cell cost/log parsing, aggregate cost reporting, pre-flight capacity check, finalize step. A clean Executor protocol needs to decide which of those are protocol-level vs executor-private.

Three design options for the protocol surface

(A) Thin protocol — submit(payload) → Future[result] only.

class Executor(Protocol):
    def submit(self, payload: CellPayload) -> Future[CellResult]: ...
    def shutdown(self) -> None: ...

Retry, cost, finalize all live in the caller (a generic dispatch() function in dispatch.py). The executor is just "how do I run one cell." Local is ThreadPoolExecutor(max_workers=1); Lambda is the boto3-fan-out; ray/dask are ray.remote / dask.delayed.

• Pro: smallest contract, easiest second/third implementation. Vaex/ray/dask all already expose submit-shaped APIs.
• Con: cost reporting becomes the caller's problem — but Lambda's per-invocation gb_seconds only comes back in the response payload, so the caller has to know to extract it. That's a leaky abstraction.

(B) Protocol carries cost + finalize hooks.

class Executor(Protocol):
    def submit(self, payload: CellPayload) -> Future[CellResult]: ...
    def preflight(self, n_cells: int) -> PreflightReport: ...
    def measure_cost(self, result: CellResult) -> CellCost: ...
    def finalize(self) -> RunReport: ...
    def shutdown(self) -> None: ...

The dispatch loop is still generic, but cost / preflight / finalize have well-defined entry points each executor implements.

• Pro: covers everything Lambda needs (preflight probe, per-cell cost from response payload, end-of-run aggregate). Ray/dask preflight is trivial (return cluster size); cost is "scheduler-reported runtime × $0".
• Con: five-method protocol — more surface to mock in tests.

(C) Protocol + retry policy as a separate strategy object.

Same as (B) but factor out retry — pass a RetryPolicy(max_attempts, backoff, classify) to dispatch() rather than each executor implementing its own retry.

• Pro: the only retry-policy variation across backends is which exceptions are retryable (errno-24 isn't; boto3 throttling is; ray scheduler timeouts depend). That's the classify callback — small, testable. Default policies live in dispatch.py (LAMBDA_RETRY, LOCAL_RETRY, etc.).
• Con: one more concept to teach.

Lean: (B) + (C). The clean Executor protocol is "submit + preflight + measure_cost + finalize," with a separate RetryPolicy object the caller hands to dispatch(). That matches what Lambda actually needs (PR #41's concurrency probe is the preflight; the cost parse is per-cell) without forcing ray/dask implementations to model the same.

Questions before I draft a plan

1. Is the local backend in scope, or just an "Executor for Lambda + a wrapper around ThreadPoolExecutor"? I'd lean in-scope — making _run_local use the same dispatch loop costs ~20 LoC and validates the protocol on the trivial case. But it does mean _run_local's behavior changes shape (becomes "trivial executor under the generic loop"). Confirm.

2. Backward compatibility on runner.agg(). Does the public agg(backend="local"|"lambda") API stay verbatim? The dispatch refactor should be invisible from outside, but I want to confirm we're not also taking the opportunity to expose agg(executor=...) for the temporal/cluster use case yet — or is that the point (we want Temporal aggregation #12 Phase 1c to be able to inject its own executor)?

3. Where does cost reporting live? Today runner.py prints gb_seconds/estimated_cost_usd. Does that stay in runner.py (the dispatch loop returns a RunReport, runner prints), or move into dispatch.py? Lean: runner owns presentation; dispatch returns structured data.

4. Concurrency-aware Lambda orchestrator: pre-flight probe + FD-exhaustion guard #41 concurrency probe. Goes behind executor.preflight() in option (B). That means concurrency.py either stays a separate helper module that LambdaExecutor.preflight() calls, or moves entirely into the executor. Lean: keep concurrency.py as helpers, call from LambdaExecutor.preflight().

5. Concurrency orchestrator: mid-run capacity re-check + grant cloudwatch:GetMetricStatistics in template.yaml #56 (mid-run capacity re-check). That's currently a separate enhancement. Is it in scope here, or does dispatch.py just provide the hook (an executor.recheck() callable invoked every N cells) and Concurrency orchestrator: mid-run capacity re-check + grant cloudwatch:GetMetricStatistics in template.yaml #56 fills it in? Lean: out of scope; this PR is the abstraction, Concurrency orchestrator: mid-run capacity re-check + grant cloudwatch:GetMetricStatistics in template.yaml #56 is a use case for it.

Once these are answered, I can write a phased plan (probably: lift _run_lambda into LambdaExecutor + dispatch() first, port _run_local, then concurrency.preflight() integration, then tests + docs).

---

Generated by Claude Code

[espg]

I'm worried that #70 got started a bit early, before this issue and #64 fully were discussed. My thinking is that we may want to restart #70 (or an effort like it) after this and #64 land for proper abstraction phasing. Agree, or is there no phasing abstraction risk?

As for the above items-- I agree with you Lean: (B) + (C) , let's do that.

1. Is the local backend in scope

Yes, in scope

2. Backward compatibility on runner.agg(). Does the public agg(backend="local"|"lambda") API stay verbatim? The dispatch refactor should be invisible from outside, but I want to confirm we're not also taking the opportunity to expose agg(executor=...) for the temporal/cluster use case yet — or is that the point (we want Temporal aggregation #12 Phase 1c to be able to inject its own executor)?

I'm not sure I understand this-- a temporal executor would still be runnable on multiple backends, i.e., backend="local"|"lambda". So what's the question here? Have a look at #64 if that helps... this issue is being done as a prereq to enabling temporal aggregations.

3. Where does cost reporting live?

Yes, runner owns presentation; dispatch returns structured data.

4. ...concurrency probe

keep concurrency.py as helpers

5. Concurrency orchestrator: mid-run capacity re-check + grant cloudwatch:GetMetricStatistics in template.yaml #56 (mid-run capacity re-check). That's currently a separate enhancement. Is it in scope here, or does dispatch.py just provide the hook (an executor.recheck() callable invoked every N cells) and Concurrency orchestrator: mid-run capacity re-check + grant cloudwatch:GetMetricStatistics in template.yaml #56 fills it in? Lean: out of scope; this PR is the abstraction, Concurrency orchestrator: mid-run capacity re-check + grant cloudwatch:GetMetricStatistics in template.yaml #56 is a use case for it.

Yeah, out of scope (this PR is the abstraction)

[espg]

🤖 from Claude

Good — (B)+(C) locked, and your answers settle 1, 3, 4, 5. Two things to close out.

Clarifying my Q2 (you flagged it didn't land)

You're right, and your framing is the answer: pipeline type and backend are orthogonal axes.

• Backend / executor = where a unit runs (local thread pool vs Lambda fan-out vs ray/dask later). This is what dispatch.py's Executor abstracts.
• Pipeline type = what a unit is (a spatial morton cell vs a temporal event). This is Temporal aggregation #12 / Plugin registry as shared vocabulary (#12, #59) #64 territory.

A temporal pipeline therefore runs on the same backend="local"|"lambda" executors — it just feeds them process_event units instead of process_shard units. So my Q2 collapses to: the public agg(backend=...) signature stays verbatim; "executor" is an internal seam selected from backend, not a new public parameter. #12 Phase 1c doesn't inject its own executor — it reuses these two behind the same backend switch. That's exactly what you described; nothing to add to the public API. Resolved.

On the #70 phasing risk (you raised the same concern on #64)

Agreed there's real risk, and for this issue it's sharpest: #70's Phase 3 dispatch.py predates this discussion — it extracted invoke_with_retry / estimate_cost / preflight_concurrency_probe as standalone functions before the (B)+(C) Executor protocol + separate RetryPolicy was locked just now. So #70's dispatch seam is the single piece most likely to need rework. (Its Phase 2 registry.py, by contrast, largely matches #64's spec and is worth keeping.)

Recommendation — rebase, don't restart:

1. Land Dispatcher abstraction as shared infrastructure (#20, #12) #63 as the canonical Executor (submit / preflight / measure_cost / finalize) + RetryPolicy(LAMBDA_RETRY, LOCAL_RETRY) in dispatch.py.
2. Rebuild Temporal aggregation infrastructure (Refs #12) #70's dispatch.py on top of that protocol; keep its registry.py.
3. Temporal aggregation infrastructure (Refs #12) #70 stays blocked on this + Plugin registry as shared vocabulary (#12, #59) #64 until then (it already is).

Restart-vs-rebase of #70 is a side-effecting call, so I won't relabel/close it — flagging the path for you.

Where this leaves #63

All five questions are answered and (B)+(C) is locked, so the design is ready to move from discuss → a phased plan / implement. Proposed phasing when you relabel:

1. Lift _run_lambda → LambdaExecutor (submit/preflight/measure_cost/finalize) behind a generic dispatch() loop; RetryPolicy extracted.
2. Port _run_local → LocalExecutor on the same loop (validates the protocol on the trivial case).
3. concurrency.py stays helpers, called from LambdaExecutor.preflight(); runner keeps cost presentation, dispatch returns the structured RunReport.
4. Tests: protocol conformance for both executors + byte-identical spatial run vs main.

Marking waiting on your nod to relabel (and the rebase-vs-restart call on #70).

---

Generated by Claude Code