robsonkades/agent-skills/distributed-aggregation-and-barriers

Distributed Aggregation And Barriers

Purpose

Get one correct, reproducible answer out of many workers, and pay as little synchronisation for it as the answer requires. Two decisions carry the whole topic: what the combining function is allowed to be, and where — if anywhere — every worker must wait for every other.

The failure this prevents is the aggregate that disagrees with itself. Same input, same code, a different partition order, and the total moves in the fifth decimal place; finance opens a reconciliation ticket nobody can reproduce, because the cause is that floating-point addition is not associative and the shuffle is not deterministic. The second failure is the barrier nobody named: a job of ten thousand tasks whose wall-clock time is set entirely by two of them, where adding workers changes nothing at all.

Workflow

Start with the affected job's registration or execution plan: locate the selected accumulator, combiner/finalizer, intermediate-state codec and output consumer. Reuse its aggregate definition, input snapshot, commit protocol and measured task profile before asking for missing context. Retain an adequate exact representation and required barrier; introduce a sketch, repartitioning or speculation only for a demonstrated constraint. Scale investigation and testing to whether the request is an explanation, design review or implementation change.

1. Write the aggregate contract. Define identity, accumulator, merge, finish, input

domain, units/window/population, overflow/error policy and whether encounter order matters. Associativity is required for arbitrary grouping; commutativity is required only when partials may be reordered. Merging separately accumulated partitions must also match accumulating their combined input under the result contract; merge algebra alone is insufficient. For keyed work, define business key equality and verify that the engine's encoding, grouping and partitioning preserve it. Neither law prevents double-counting a repeated attempt.

1. Rewrite aggregates that lack a mergeable sufficient state. Average becomes a

(sum, count) pair; variance becomes (n, mean, M2); a pooled percentile needs a mergeable distribution summary. A ratio needs a denominator with the correct exposure semantics.

1. Choose a summary per metric and state its error. Exact where cardinality is small, an

approximate mergeable sketch when its error is acceptable and exact state is too costly. Make that error part of the consumer's result contract.

1. Partition by measured cost, not merely count, when skew explains stragglers; distinguish

deterministic data skew from host faults or transient resource contention.

1. Place the barriers deliberately and count them. Every barrier converts the slowest

participant into everyone's latency. Bind completion to the expected logical partitions and their committed outputs in one epoch; worker replacement must preserve input coverage. Ask what breaks if this barrier is removed.

1. Define attempt identity and output commit. Every logical partition may execute more

than once. Stage output by (job, stage, partition, attempt) and atomically select one successful attempt, or use a sink-specific idempotent/transactional commit protocol.

1. Decide the partial-failure contract before the job runs, not during the incident:

fail the job, retry the failed tasks, or emit a partial result with an explicit completeness record.

1. Check algebra and recovery. Property-test regrouping/reordering allowed by the

contract, including accumulator/merge compatibility and serialized-state round trips. Verify that tests exercise the selected merge/codec hooks, not only a sequential helper. If precombination is optional or repeatable, test those invocation shapes too; correctness must not depend on an optional hook running exactly once. Test equivalent and distinct keys across workers. Inject duplicate attempts and crashes at commit boundaries, and compare against a trusted sequential oracle. Report the schedules and faults actually tested; finite trials support the contract but do not prove every distributed execution safe.

Inspect engine and sketch-library versions, input snapshot, numeric domain and sink commit guarantees; for Java, inspect the target JDK/toolchain too. Reference records require JDK 16+; test sketches assume project-specific JUnit/AssertJ fixtures and are not standalone programs. Do not upgrade the target to fit an example. Deliver the aggregate/equivalence contract, evidence for merge and recovery semantics, expected participant/completeness record and remaining validation gaps.

Decision block

Use a barrier when:
- a later stage genuinely reads the complete output of an earlier one — a global sort, a
  normalisation by a total, a join needing both sides fully partitioned
- the participant set is bounded and known before the stage starts
The barrier is affordable when:
- measured max-stage latency, not merely p99/p50, fits the job SLO at the actual task count
Avoid a barrier when:
- incremental consumption preserves the required semantics and waiting adds unnecessary
  exposure to stragglers; retain a required completeness gate despite a long tail
- participants can join or fail mid-stage and no epoch/membership protocol defines who
  counts as a participant
- the downstream stage could consume results incrementally instead
Prefer incremental or hierarchical combination instead when the combining function is
  associative under the permitted grouping and order (commutative if reordered). Combining
  early does not remove the participant/completeness condition for a final batch result.
  Continuously read results and window semantics belong to streaming-pipeline-topologies.
Speculatively re-execute a straggler only when attempts satisfy the same result contract,
  one output is selected, and external effects are absent or independently safe across all
  attempts and late effects (idempotency). Cap copies and account for losers still running.

Rules