Parallel fan-out of multi-bucket downsampling tasks
A multi-region CDN edge-metrics platform lands every point-of-presence’s request logs into one validated bucket, edge_metrics_validated, after a quality gate has stamped them. From there, four regions — us-east, eu-west, ap-south, sa-east — each need their own five-minute downsampling pass, and a single daily rollup must not begin until all four have finished, or it will average an incomplete picture of the world. Running the four passes one after another wastes most of the hourly budget waiting on serial I/O, but firing the daily rollup on a fixed clock offset instead of an actual completion signal is worse: it produces a daily point that silently omits whichever region was still writing. This page builds the fan-out/fan-in shape correctly — parallel per-region branches, an explicit barrier that waits on real run status, and a downstream rollup that only starts once the barrier is satisfied. It is the concrete parallel case of the dependency modelling covered in Dependency Mapping & DAG Construction.
Prerequisites
Solution walkthrough
1. Model the fan-out/fan-in DAG
Draw the shape before writing code, because the shape dictates where the barrier goes. One upstream node fans out to N sibling branches that share no data dependency on each other; every branch must complete before a single downstream node — the join — is allowed to run. The parallel branches collapse the wall-clock cost from the sum of branch durations to roughly the slowest branch plus the join, which is exactly why the pattern is worth the coordination:
$$ T_{\text{parallel}} \approx \max_{i}, t_i + T_{\text{join}} \qquad\text{versus}\qquad T_{\text{serial}} = \sum_{i} t_i $$
With four regions each taking about three minutes, the serial path burns roughly twelve minutes of the hourly window while the parallel path finishes in about four. The catch is the join: it is only a barrier if it waits on the actual success of all four branches, not on elapsed time.
2. Template the per-region downsampling task
Each branch is the same Flux logic parameterised by region, so the only difference between the four tasks is a single string. Keeping them separate — rather than one task looping over regions — means a failure in ap-south cannot poison the other three, and each branch can be retried in isolation.
option task = {name: "downsample-edge-us-east", every: 1h, offset: 3m}
region = "us-east"
from(bucket: "edge_metrics_validated")
|> range(start: -task.every)
|> filter(fn: (r) => r.region == region)
|> filter(fn: (r) => r._measurement == "edge_request")
|> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
|> set(key: "region", value: region)
|> to(bucket: "edge_metrics_5m")
The filter(fn: (r) => r.region == region) line is what makes the branches independent: each reads only its own slice of the shared source. The explicit set(key: "region", ...) re-stamps the region tag onto the aggregate so that four branches writing to one shared edge_metrics_5m target land on disjoint series instead of overwriting one another. offset: 3m gives late edge batches a moment to arrive before the window closes; keep it smaller than the orchestrator’s per-branch timeout. Because these tasks are triggered by the orchestrator, their every value is a declaration of the window they process, not the clock that fires them.
3. Launch the parallel branches with a barrier
The orchestrator triggers all four region tasks at once, then blocks on a barrier that resolves only when every branch reports success. asyncio.gather is the barrier primitive here; the synchronous InfluxDB client calls are pushed onto worker threads with asyncio.to_thread so the event loop stays free to poll all four branches concurrently. This is the same reusable-client and concurrency discipline described in Python client orchestration patterns.
import asyncio
import os
import time
from influxdb_client import InfluxDBClient
REGIONS = ["us-east", "eu-west", "ap-south", "sa-east"]
POLL_SECONDS = 5
RUN_TIMEOUT = 900 # 15 min hard ceiling per branch
client = InfluxDBClient(
url=os.environ["INFLUX_URL"],
token=os.environ["INFLUX_TOKEN"],
org=os.environ["INFLUX_ORG"],
)
tasks_api = client.tasks_api()
def task_id(name: str) -> str:
matches = tasks_api.find_tasks(name=name)
if not matches:
raise RuntimeError(f"no task named {name!r}")
return matches[0].id
async def run_branch(region: str) -> str:
tid = task_id(f"downsample-edge-{region}")
run = await asyncio.to_thread(tasks_api.run_manually, tid)
deadline = time.monotonic() + RUN_TIMEOUT
while time.monotonic() < deadline:
current = await asyncio.to_thread(tasks_api.get_run, tid, run.id)
if current.status == "success":
return region
if current.status in ("failed", "canceled"):
raise RuntimeError(f"{region} branch {current.status}")
await asyncio.sleep(POLL_SECONDS)
raise TimeoutError(f"{region} branch exceeded {RUN_TIMEOUT}s")
async def fan_out_fan_in() -> None:
results = await asyncio.gather(
*(run_branch(r) for r in REGIONS), return_exceptions=True
)
failed = [
(r, res) for r, res in zip(REGIONS, results) if isinstance(res, Exception)
]
if failed:
# barrier NOT satisfied — do not release the join
raise RuntimeError(f"barrier failed on: {failed}")
# every branch succeeded — the 5m tier is complete for the window
daily = task_id("daily-rollup-edge")
await asyncio.to_thread(tasks_api.run_manually, daily)
if __name__ == "__main__":
try:
asyncio.run(fan_out_fan_in())
finally:
client.close()
return_exceptions=True is deliberate: it lets all four branches run to completion so you learn every failing region in one pass rather than aborting on the first, while the subsequent failed check still refuses to release the join unless the list is empty. The per-branch RUN_TIMEOUT guarantees the barrier cannot hang forever on a wedged region. This branch-then-join structure is the runnable form of the graphs built in building dependency graphs for multi-stage pipeline execution.
4. Trigger the daily fan-in rollup after the barrier
The join task reads the shared five-minute tier that all four branches have now populated and collapses it to one daily point per region-series. Because it is only ever fired by fan_out_fan_in after the barrier passes, it never sees a partial window.
option task = {name: "daily-rollup-edge", every: 24h, offset: 20m}
from(bucket: "edge_metrics_5m")
|> range(start: -task.every)
|> filter(fn: (r) => r._measurement == "edge_request")
|> group(columns: ["region", "_measurement", "_field"])
|> aggregateWindow(every: 24h, fn: mean, createEmpty: false)
|> to(bucket: "edge_metrics_daily")
Grouping by region before the daily aggregateWindow preserves one series per region rather than blending all four into a single global mean — the join combines the branches without erasing which region each value came from. If your daily metric is a sum rather than a mean, deriving it from the five-minute tier avoids re-scanning raw data, but watch the mean-of-means trap; the count-weighted approach for multi-tier chains is covered in Multi-Tier Rollup Scheduling.
Gotchas and edge cases
A time-based barrier is not a barrier. The single most common failure is scheduling daily-rollup-edge at, say, offset: 20m and trusting that the region tasks “usually finish by then”. The first time sa-east runs long — a compaction, a slow query, a retried batch — the daily point for that day is computed from three regions, not four, and nothing errors. Only a barrier that polls real run status, as in Step 3, can guarantee completeness; keep the daily task off any wall-clock schedule and let the orchestrator own its trigger.
Fan-out branches must write disjoint series. Four branches writing to one edge_metrics_5m bucket are safe only because each re-stamps its own region tag. Drop the set(key: "region", ...) and every branch writes the same tag-set for the same timestamps, so the last writer silently clobbers the others and three regions’ work vanishes. Any shared write target in a fan-out must be partitioned by a tag the branches cannot collide on.
One slow region should degrade, not deadlock. With return_exceptions=True a failed branch surfaces cleanly, but decide explicitly what a partial fleet means for your rollup. Refusing to fire the daily job (the default above) is correct when completeness is non-negotiable; if a delayed region is acceptable, record which regions were present and fire a clearly-flagged degraded rollup rather than silently pretending the window was whole. Either way the decision is explicit and logged — never an accident of timing.
Verification
Before trusting the daily point, confirm the five-minute tier actually holds all four regions for the window the barrier just cleared. A distinct-region count equal to the fleet size proves the fan-in was complete:
from(bucket: "edge_metrics_5m")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "edge_request")
|> keep(columns: ["region"])
|> group()
|> distinct(column: "region")
|> count()
|> yield(name: "regions_present_should_be_4")
A result of 4 confirms every branch landed data before the join released; anything less means the barrier let a partial window through and the daily rollup for that hour should be recomputed once the missing region catches up.
FAQ
Why run four separate tasks instead of one task that loops over regions?
Isolation. Separate tasks let a single region fail, retry, and be reasoned about independently, and they let the orchestrator poll each branch’s status distinctly for the barrier. One looping task collapses all four into a single success/failure verdict, so a fault in one region takes down the entire window and you lose the fine-grained run status the fan-in decision depends on.
Can I use the task scheduler’s own offset as the barrier?
No. An offset only shifts when a task starts relative to its window boundary; it carries no knowledge of whether upstream branches have finished. Two tasks scheduled close together can still overlap or reorder under load. A real barrier must observe completion, which is why Step 3 polls run status rather than trusting the clock.
How does this scale beyond four regions?
asyncio.gather and return_exceptions=True scale to dozens of branches without code changes, but at large fan-out add a concurrency limit (an asyncio.Semaphore) so you do not trigger hundreds of task runs at once and overwhelm the server. The barrier logic is unchanged; only the launch is throttled.
Related
- Dependency Mapping & DAG Construction — the parent guide covering how fan-out/fan-in and other execution shapes are modelled as directed graphs.
- Python Client Orchestration Patterns — reusable-client, async, and retry patterns that harden the orchestrator loop above.
- Multi-Tier Rollup Scheduling — deriving daily points from an intermediate tier without double-counting once the join releases.
Up one level: Dependency Mapping & DAG Construction