feat(router): engine flags for dataflow execution and execution plan scheduling

router-tests/bench/engine-modes/README.md

@@ -0,0 +1,95 @@
+# Engine-modes benchmark suite
+
+Measures the router's three engine execution modes against each other from ONE
+binary, selected by env flags:
+
+| label | `ENGINE_ENABLE_DATAFLOW` | `ENGINE_ENABLE_SCHEDULE_TREE` |
+|---|---|---|
+| cosmo-baseline | false | false |
+| cosmo-dataflow | true | false |
+| cosmo-scheduler | false | true |
+
+The suite is self-contained and cosmo-only.
+It exists to prove two properties on every change to the engine execution path:
+
+1. The flags cost NOTHING when latency is uniform (regression gate).
+2. Under skewed subgraph latency the dataflow executor and the schedule-tree
+   scheduler collapse the median to the dependency-graph critical path
+   (capability gate; reference: −37% vs baseline on the workload below).
+
+## Provisioning
+
+1. Router image (from this repo):
+
+```bash
+cd router && CGO_ENABLED=0 GOOS=linux GOARCH=$(go env GOARCH) go build -ldflags '-s -w' -o /tmp/bench-router/cosmo ./cmd/router
+printf 'FROM alpine:3.20\nRUN apk add --no-cache ca-certificates\nWORKDIR /app\nCOPY cosmo /app/cosmo\nENTRYPOINT ["/app/cosmo"]\n' > /tmp/bench-router/Dockerfile
+docker build -t bench-cosmo:local /tmp/bench-router
+```
+
+2. Subgraphs image: any implementation of the standard federation example
+   subgraphs (accounts/products/reviews/inventory on one port, 4200), built as
+   `bench-subgraphs:latest`. The reference dataset used for the acceptance
+   numbers is the one from the graphql-hive/gateways-benchmark repository
+   (workload source only — this suite benchmarks no other gateways).
+   The canonical federated query (`QUERY_FILE`) is the benchmark's `MyQuery`;
+   against this dataset the full response is exactly 87,599 bytes
+   (`EXPECTED_BYTES`).
+
+3. lat-proxy (skew scenario only):
+
+```bash
+docker build -t lat-proxy:local lat-proxy/
+```
+
+4. Router config dirs (execution config + minimal yaml), one per topology:
+   `COSMO_CFG_DIR` for the uniform scenario routes subgraphs to
+   `http://bench-sg:4200/<name>`, the skew variant to
+   `http://lat-proxy:8080/<name>`.
+
+## Run
+
+```bash
+export QUERY_FILE=/path/to/query.json EXPECTED_BYTES=87599
+export COSMO_CFG_DIR=/path/to/cfg ROUTER_IMAGE=bench-cosmo:local
+
+# Gate first — abort on any divergence:
+bash byteid.sh
+
+# Then the matrices (macOS: under caffeinate -dimsu):
+RES=/tmp/results/uniform.jsonl bash uniform_matrix.sh
+COSMO_CFG_DIR=/path/to/cfg-proxy RES=/tmp/results/skew.jsonl bash skew_matrix.sh
+```
+
+Each run appends one JSON line: rps, success, min/med/p90/p95/p99.9, cpu
+avg/max, mem max, valid_bytes, label, flags. Aggregate as median-of-3 per
+(scenario, label) cell; a cell whose rep RPS spread exceeds 1.1x is re-run.
+
+## Acceptance thresholds
+
+Reference machine: Apple M4 Max, OrbStack, gateway pinned to 4 CPUs / 2 GB.
+
+- Byte identity: `BYTEID GATE: PASS` — one sha256 across all 4 modes
+  (baseline/dataflow/scheduler/both), 50 requests each, exact `EXPECTED_BYTES`.
+- Skew (`accounts=20,products=20,reviews=20,inventory=150`; critical path
+  210 ms): dataflow AND scheduler median ≤ 225 ms at 1 VU and 64 VU, RPS ≥ 265
+  at 64 VU (reference floors: 218–219 ms / 274–275 RPS); baseline median in the
+  340–360 ms band (the barrier anchor). Both new modes ≈ −37% vs baseline.
+- Uniform ({0,10,50,100} ms, 64 VU): all three modes within 3% of each other on
+  median latency AND RPS at every delay; at 0 ms ≥ 19k RPS under conditions
+  that reproduce the reference anchors (absolute RPS is machine-sensitive —
+  when anchors don't reproduce, only the within-3% relative gate applies).
+- `cosmo-both-SMOKE` rows ≈ the scheduler row (safe-degradation proof).
+
+## Methodology notes (encoded in the scripts)
+
+- netem on the SUBGRAPH container's egress with `limit 100000`, qdisc verified
+  present after adding; gateway CPU-pinned via `--cpuset-cpus` with
+  `GOMAXPROCS`/`GOMEMLIMIT`; subgraphs unpinned; k6 on the same docker network
+  (no host NAT); closed-loop constant VUs.
+- A run starts only after a real federated response of the expected size is
+  observed; `valid_bytes` is recorded per row.
+- k6 `setup()` warmup is capped at 8 requests (`vus*2` sequential warmup
+  poisons `http_reqs.rate` windowing at high latency).
+- Tail latencies ≥ 10x the injected delay are the macOS host-sleep signature —
+  invalidate the rep.

router-tests/bench/engine-modes/byteid.sh

@@ -0,0 +1,111 @@
+#!/usr/bin/env bash
+# Byte-identity gate: the SAME federated query must return the SAME bytes from
+# ONE router image across all engine-flag modes, 25 sequential + 25 concurrent
+# requests each. Aborts the benchmark protocol on any divergence.
+#
+# Modes: baseline(off,off) dataflow(on,off) scheduler(off,on) both(on,on).
+# "both" asserts safe degradation: schedule-tree plans are nested, the dataflow
+# executor structurally rejects nested plans and falls back — same bytes.
+#
+# Plumb-through preflight per mode (LOG_LEVEL=info): when dataflow is on the
+# logs must contain "dataflow executor enabled"; when the scheduler is on,
+# "schedule-tree planner enabled" — proving env -> config -> engine wiring.
+#
+# Success-path only by design: multi-error ordering determinism is covered
+# in-engine by TestDataflowErrorOrderDeterminism.
+set -uo pipefail
+ROUTER_IMAGE="${ROUTER_IMAGE:-bench-cosmo:local}"
+SUBGRAPHS_IMAGE="${SUBGRAPHS_IMAGE:-bench-subgraphs:latest}"
+COSMO_CFG_DIR="${COSMO_CFG_DIR:?set COSMO_CFG_DIR (subgraph URLs -> http://bench-sg:4200)}"
+QUERY_FILE="${QUERY_FILE:?set QUERY_FILE}"
+EXPECTED_BYTES="${EXPECTED_BYTES:-87599}"
+NET=bench-net; SG=bench-sg; GWC=bench-gw
+WORK="${WORK:-$(mktemp -d)}"
+
+cleanup(){ docker rm -f "$GWC" "$SG" >/dev/null 2>&1; }
+cleanup
+docker network rm "$NET" >/dev/null 2>&1; sleep 1; docker network create "$NET" >/dev/null 2>&1
+
+docker run -d --name "$SG" --network "$NET" "$SUBGRAPHS_IMAGE" >/dev/null 2>&1
+for i in $(seq 1 60); do
+  docker run --rm --network "$NET" curlimages/curl:latest -fsS --max-time 1 -X POST http://$SG:4200/products -H 'content-type: application/json' -d '{"query":"{topProducts{upc}}"}' >/dev/null 2>&1 && break
+  sleep 0.5
+done
+
+BASE_HASH=""
+FAIL=0
+
+run_mode(){
+  local mode="$1" dataflow="$2" schedule="$3"
+  docker rm -f "$GWC" >/dev/null 2>&1
+  docker run -d --name "$GWC" --network "$NET" \
+    -e LISTEN_ADDR=0.0.0.0:4000 -e LOG_LEVEL=info \
+    -e TRACING_ENABLED=false -e METRICS_ENABLED=false -e METRICS_OTLP_ENABLED=false \
+    -e GRAPHQL_METRICS_ENABLED=false -e PROMETHEUS_ENABLED=false -e ACCESS_LOGS_ENABLED=false \
+    -e ENGINE_ENABLE_DATAFLOW="$dataflow" -e ENGINE_ENABLE_SCHEDULE_TREE="$schedule" \
+    -e ROUTER_CONFIG_PATH=/etc/cosmo/config.json -e CONFIG_PATH=/etc/cosmo/config.yaml \
+    -v "$COSMO_CFG_DIR:/etc/cosmo:ro" "$ROUTER_IMAGE" >/dev/null 2>&1
+
+  local EP="http://$GWC:4000/graphql" up=0
+  for i in $(seq 1 120); do
+    docker run --rm --network "$NET" curlimages/curl:latest -fsS --max-time 3 -X POST "$EP" -H 'content-type: application/json' -d '{"query":"{__typename}"}' >/dev/null 2>&1 && { up=1; break; }
+    sleep 0.5
+  done
+  [ "$up" = 1 ] || { echo "BYTEID FAIL [$mode]: router did not come up"; docker logs "$GWC" 2>&1 | tail -5; FAIL=1; return; }
+
+  local logs; logs=$(docker logs "$GWC" 2>&1)
+  if [ "$dataflow" = "true" ] && ! echo "$logs" | grep -q "dataflow executor enabled"; then
+    echo "BYTEID FAIL [$mode]: ENGINE_ENABLE_DATAFLOW not plumbed through"; FAIL=1; return
+  fi
+  if [ "$schedule" = "true" ] && ! echo "$logs" | grep -q "schedule-tree planner enabled"; then
+    echo "BYTEID FAIL [$mode]: ENGINE_ENABLE_SCHEDULE_TREE not plumbed through"; FAIL=1; return
+  fi
+
+  local outdir="$WORK/byteid_$mode"
+  rm -rf "$outdir"; mkdir -p "$outdir"
+  for i in $(seq 1 25); do
+    docker run --rm --network "$NET" -v "$QUERY_FILE:/q.json:ro" curlimages/curl:latest \
+      -fsS --max-time 30 -X POST "$EP" -H 'content-type: application/json' --data-binary @/q.json \
+      >"$outdir/seq_$i.json" 2>/dev/null
+  done
+  # 25 concurrent. Helper script: the redirect must live INSIDE the spawned
+  # shell, and BSD xargs rejects long inline -I{} commands.
+  cat > "$WORK/byteid_curl.sh" <<HELPER
+#!/bin/sh
+docker run --rm --network $NET -v $QUERY_FILE:/q.json:ro curlimages/curl:latest -fsS --max-time 30 -X POST $EP -H 'content-type: application/json' --data-binary @/q.json > $outdir/conc_\$1.json 2>/dev/null
+HELPER
+  chmod +x "$WORK/byteid_curl.sh"
+  seq 1 25 | xargs -P 25 -n 1 "$WORK/byteid_curl.sh"
+
+  local hashes bytes nfiles
+  hashes=$(shasum -a 256 "$outdir"/*.json | awk '{print $1}' | sort -u)
+  bytes=$(wc -c < "$outdir/seq_1.json" | tr -d ' ')
+  nfiles=$(ls "$outdir"/*.json | wc -l | tr -d ' ')
+  # The gate is 25 sequential + 25 concurrent; a partial concurrent batch
+  # (xargs failure, curl errors) must not pass silently.
+  if [ "$nfiles" != 50 ]; then
+    echo "BYTEID FAIL [$mode]: expected 50 response files, got $nfiles"; FAIL=1; return
+  fi
+  local nhashes; nhashes=$(echo "$hashes" | grep -c .)
+  if [ "$nhashes" != 1 ]; then
+    echo "BYTEID FAIL [$mode]: $nhashes distinct hashes within mode ($nfiles files)"; FAIL=1; return
+  fi
+  if [ "$bytes" != "$EXPECTED_BYTES" ]; then
+    echo "BYTEID FAIL [$mode]: byte length $bytes != $EXPECTED_BYTES"; FAIL=1; return
+  fi
+  if [ -z "$BASE_HASH" ]; then
+    BASE_HASH="$hashes"
+  elif [ "$hashes" != "$BASE_HASH" ]; then
+    echo "BYTEID FAIL [$mode]: hash differs from baseline"; FAIL=1; return
+  fi
+  echo "BYTEID OK [$mode]: $nfiles/$nfiles identical, sha256=${hashes:0:16}..., bytes=$bytes"
+}
+
+run_mode baseline  false false
+run_mode dataflow  true  false
+run_mode scheduler false true
+run_mode both      true  true
+
+cleanup; docker network rm "$NET" >/dev/null 2>&1
+if [ "$FAIL" != 0 ]; then echo "BYTEID GATE: FAIL"; exit 1; fi
+echo "BYTEID GATE: PASS (all 4 modes byte-identical, $EXPECTED_BYTES bytes)"