# Scatter–gather (fan-out / fan-in) with `resolve_callback_worker` This example shows how to answer **one** request by fanning it out into **many** parallel sub-tasks and aggregating their results into a single response — a map-reduce in miniature. The clever part is *where* the aggregation happens. Every event a worker sends back arrives at the manager's `resolve_callback_worker`. By overriding it, the manager becomes the **gather** point: it collects partial results and, only when the last one arrives, completes the original request. * the manager receives one API request * it **scatters** N sub-tasks to N workers — and deliberately does **not** set a response * each worker computes its part and emits it back * `resolve_callback_worker` **gathers** the parts; on the last one it fills the original request's response and sends it * the client gets a single aggregated answer The toy task sums the squares `1² + 2² + … + n²`, one worker per term. Replace the worker body with whatever you actually fan out (querying N shards, calling N upstreams, processing N file chunks). --- ## Why this works Two framework mechanics make the pattern possible: 1. **A handler that sets no response is not auto-answered.** After a `@request` handler runs, the manager only sends the event back if the handler *changed* `event.response`. The scatter handler leaves the response untouched, so the request simply stays pending at the Broker — there is nothing to answer yet. 2. **The original event still carries its correlation identity.** Keep a reference to it; when the last part arrives, set its response and put it on the router queue. It is the same object the Broker is waiting on, so it resolves *that* client's request — even with many requests in flight. `WorkerQuit` (sent automatically when each worker finishes) is handled by the framework *before* `resolve_callback_worker`, so your override only ever sees the partial-result events. Call `super().resolve_callback_worker(...)` for anything you don't recognize. --- ## Events ```python # events/squares.py from http import HTTPMethod from edri.api.dataclass.api_event import api, ApiType from edri.dataclass.event import Event, event from edri.dataclass.response import Response, response @response class SumResponse(Response): total: int parts: int @api(url="/sum-of-squares/{n}", method=HTTPMethod.GET, resource="sum-of-squares", exclude={ApiType.HTML, ApiType.WS}) class SumOfSquares(Event): n: int response: SumResponse # Internal carrier: one sub-task in, the same event (with result) back. @event class SquarePart(Event): job_id: str # correlates the part to its scatter job value: int # input result: int = 0 # filled by the worker ``` --- ## Worker Each worker computes a single term and sends its `SquarePart` back. A worker **must** call `event_send(...)` — modifying the event is not enough. ```python # workers/squares.py from edri.abstract.worker import WorkerProcess from events.squares import SquarePart class SquareWorker(WorkerProcess[SquarePart]): def do(self) -> None: self.event.result = self.event.value * self.event.value self.event_send(self.event) ``` --- ## Manager (scatter + gather) ```python # managers/squares.py from uuid import uuid4 from edri.abstract import ManagerBase, request from edri.dataclass.event import Event from edri.abstract.worker import Worker from events.squares import SumOfSquares, SquarePart from workers.squares import SquareWorker class FanOutManager(ManagerBase): def after_start(self) -> None: # job_id -> {"expected": int, "parts": list[int], "origin": SumOfSquares} self._jobs: dict = {} @request def sum_of_squares(self, event: SumOfSquares) -> None: # SCATTER: one worker per term, keyed by a fresh job id. job_id = uuid4().hex values = list(range(1, event.n + 1)) self._jobs[job_id] = {"expected": len(values), "parts": [], "origin": event} for value in values: self.workers.start(SquarePart(job_id=job_id, value=value), SquareWorker) # Deliberately NO response here -> the request stays pending until gather. def resolve_callback_worker(self, event: Event, worker: Worker) -> None: # GATHER: collect parts; complete the original request on the last one. if isinstance(event, SquarePart): job = self._jobs.get(event.job_id) if job is None: return # unknown / already-completed job job["parts"].append(event.result) if len(job["parts"]) == job["expected"]: self._jobs.pop(event.job_id) origin = job["origin"] origin.response.total = sum(job["parts"]) origin.response.parts = len(job["parts"]) self.router_queue.put(origin) # resolves the waiting client return # Anything else (e.g. results of other patterns) keeps default behavior. super().resolve_callback_worker(event, worker) ``` To cap how many sub-tasks run at once, give the manager a worker limit — `FanOutManager(max_workers=8)`. The framework queues the rest and drains them as workers finish; the gather logic is unchanged. --- ## Entry point ```python # main.py from edri import EDRI from managers.squares import FanOutManager if __name__ == "__main__": app = EDRI() app.add_component(FanOutManager()) app.start_api() app.run() ``` --- ## Calling it ```bash curl -H "Accept: application/json" http://localhost:8080/sum-of-squares/5 # {"data": {"total": 55, "parts": 5}, "status": "OK"} curl -H "Accept: application/json" http://localhost:8080/sum-of-squares/10 # {"data": {"total": 385, "parts": 10}, "status": "OK"} ``` --- ## Things to watch * **Response timeout.** All parts must arrive within `EDRI_API_RESPONSE_TIMEOUT` (default 60 s), or the Broker gives up on the client even though the job eventually finishes. Size the fan-out accordingly. * **A part that never arrives.** If a worker dies without sending its result, the job count never reaches `expected` and the entry leaks. For production, also count `WorkerQuit` (override the manager's worker-removal path) or attach a Scheduler-driven timeout that fails the job and answers with an error status. * **CPU vs I/O.** `WorkerProcess` gives true parallelism for CPU-bound parts; for I/O-bound fan-out (N network calls) prefer `WorkerThread`. See {doc}`/worker`.