Question
You control `N` workers and one coordinator. Worker `i` can read only shard `i`. Endpoints communicate by messages; there is no shared mutable dataset. Sending captures an independent snapshot of the payload; later sender edits cannot change an already-sent message. `ctx.read(n)` consumes up to `n` items, where `n` is a non-negative integer; `ctx.read()` consumes the remaining items. Returned lists are independent copies. ```python def worker(ctx: WorkerContext, task: str) -> None: ... def coordinator(ctx: CoordinatorContext, task: str) -> int | float: ... ``` The task is either `mode` or `median`. For mode, return the globally most frequent integer with the smaller value winning ties. For median, return the exact middle value as a float, or the mean of the two middle values for even cardinality. The global input is non-empty; individual shards may be empty. ## Requirements - Define message shapes and an explicit completion condition. - Do not assume the coordinator can read worker data directly. - Avoid shipping complete raw shards in the scalable solution. - Keep worker and coordinator receive counts consistent so the protocol terminates. - Preserve correctness under duplicate values, negative values, ties, and uneven shard sizes. ## Staged follow-ups 1. Establish a correct baseline and report its communication cost. 2. Replace centralized mode aggregation with key ownership and partial-count reduction. 3. Implement exact distributed median with repeated rank queries or range partitioning. 4. Minimize either bytes or round trips under a stated cost model. 5. Add timeouts, retry-safe messages, and recovery from a lost worker. 6. Secure and version the wire protocol for a multi-host deployment.