feat(tools/e2e): add limits harness two-tier topology composer (#381)

progress.txt

@@ -8068,3 +8068,85 @@ proving each axis's binding constraint is computed independently of the others.
 controller's (in the "Synthetic load driver (#378)" section, which now covers all four
 deep modules) and a matching entry in Layout; the topology composer remains the one
 not-yet-implemented piece of #323.
+
+### #381 - Limits harness: two-tier topology composer (aggregating Shipper stack)
+
+- Added `tools/e2e/scripts/run_limits_two_tier.sh`, a new sibling E2E scenario (podman,
+  no compose, same cleanup/`DC_E2E_KEEP` convention as `run.sh`/`run_degraded.sh`) that
+  brings up: a shared Postgres + RustFS on a dedicated bridge network (container-name
+  DNS, `run_degraded.sh`'s own convention, not `--network host` — N real DC stacks all
+  binding the default Bridge-to-Shipper port would collide on a shared host netns); an
+  **aggregating Shipper** — a standalone Vector instance, no dc_bridge/ROS involved, with
+  a `type = "vector"` source (tier-1 ingress) and a `type = "fluent"` source (synthetic-
+  sender ingress) both feeding a `postgres` sink into the real `dc_records` table; **N
+  real DC stacks** (`params/e2e_limits_params.yaml`, each its own dc_bringup + dc_bridge
+  + local Shipper); and **M synthetic senders** (one `scripts/load_driver.py` invocation
+  against the aggregator's fluent port, #378). At a fixed, non-saturating load it reuses
+  `verify_zero_loss.py` **unmodified** against one representative real stack's ledger.
+- The two-tier mechanism: each real stack's own local Shipper has no Destination that
+  writes Records to the real Postgres. `e2e_limits_params.yaml` replaces the usual
+  `pgsql_records` with `local_sink`, a throwaway `type: file` Destination whose only job
+  is to make dc_bridge render the `dc.<tag>` route branches those topics need (a
+  Destination's `inputs` is the only thing that creates one — see `render.cpp`).
+  `params/e2e_limits_forward_sink.toml` — an ADR-0003 `custom_config_files` passthrough,
+  the same mechanism `e2e_passthrough_sink.toml`/`e2e_mcap_sink.toml` already use — wires
+  a `type = "vector"` sink to those same route branches, relaying them over the network
+  to the aggregating Shipper via Vector's own native inter-instance relay protocol (the
+  standard "agent Vector forwards to aggregator Vector" pattern; **not** a `fluent` sink,
+  which Vector doesn't have — confirmed against Vector's own sink reference before
+  writing any config). Verified directly against the pinned Vector build before writing
+  the ROS-side config at all: a standalone `fluent` source -> `vector` sink -> `vector`
+  source -> `file` sink chain relayed `load_driver.py` frames end to end with every
+  custom field intact, including the ack round trip. `pgsql_files`/`rustfs` (Files) point
+  straight at the real shared Postgres/RustFS instead of through the chain — see the
+  first discovery below.
+- **Verified end-to-end for real**, not just read through: podman was available in this
+  sandbox with the images already cached, so this was actually run — a `dc-e2e:latest`
+  image was rebuilt (`Containerfile.e2e` only, on top of the existing cached
+  `dc-workspace:latest` base — no colcon recompile needed, since this ticket touches no
+  dc_bridge/measurement_server C++) and the full scenario executed repeatedly against it.
+  Found and fixed two real bugs this way before it passed clean: (1) the
+  `measure_rtt.py`-based readiness polls (copied from `run_degraded.sh`'s own pattern)
+  never override `$DC_IMAGE`'s `ENTRYPOINT` (`entrypoint.sh`, which launches the full ROS
+  stack and appends any extra args to that launch command rather than running them
+  standalone) — replaced with a plain `--entrypoint bash` + `/dev/tcp` probe, which also
+  drops the dependency on `measure_rtt.py` happening to be baked into whatever image is
+  passed in; (2) `aws-cli` 2.36's `--endpoint-url` parser rejects a hostname containing
+  an underscore outright — this repo's established `dc_e2e_*` container-naming
+  convention (used unmodified by `run_degraded.sh` too) trips it, so this script's own
+  container/network names use hyphens (`dc-e2e-limits-*`) instead; volume names, never
+  used as a hostname, keep the usual underscored convention. With both fixed, a full run
+  (`REAL_STACKS=1`, default `SYNTH_CONNECTIONS=50`/`RATE_HZ=5`, default 30s steady state
+  + 15s drain) against the real image passed clean: `ZERO-LOSS VERIFICATION PASSED (25
+  sources checked, 0 violations)`.
+- **Two discoveries recorded** (in the script's own header and the README, not worked
+  around), matching this ticket's own "if that assumption turns out false, record it"
+  hedge on its no-new-DC-code acceptance criterion:
+  1. Files bypass the two-tier chain entirely. The Uploader — File bytes over its own AWS
+     SDK client, plus its own file-metadata Records — talks directly to whatever
+     `s3`/`postgres` Destination host is configured, with no Vector hop at all
+     (`bridge_node.cpp`'s `run_uploader_worker` never touches the rendered Vector
+     config); there is no way to route Files through the aggregating Shipper without new
+     DC code. Extending true two-tier coverage to Files is left as a follow-up.
+  2. The aggregating Shipper's one `postgres` sink is a shared bottleneck no single-tier
+     scenario has (each of those stacks writes straight to its own Postgres connection).
+     Verified empirically: two real stacks (`REAL_STACKS=2`, with and without a 5-10s
+     startup stagger between them) reliably produced far more at-least-once re-delivery
+     than one does — a direct Postgres query confirmed the "duplicates" are identical
+     `(date, value)` pairs, genuine at-least-once repeats, not loss — but enough of them
+     to trip `verify_zero_loss.py`'s zero-tolerance memory-timestamp-uniqueness check
+     (`check_timestamp_resolution`), which every other check in that script tolerates as
+     a note. `DC_E2E_LIMITS_REAL_STACKS` stays configurable (not hardcoded) since finding
+     exactly where that bottleneck sits is the ramp controller's job (#323's next piece),
+     not this composer's — but its default is 1 real stack so the scenario this ticket
+     asks for passes reliably on its own, matching #381's explicit "before any ramping
+     logic exists, at a fixed, non-saturating load" scope. Also initially misdiagnosed as
+     a `MeasurementServer::setRunId()` lifecycle race (a real, separate, pre-existing
+     `$HOME/run_id` counter-file behavior with no locking) before a staggered-startup
+     test ruled that theory out — noted here so a future investigation doesn't retread
+     that path.
+- `tools/e2e/README.md` gained a "Two-tier topology composer (#381)" section (after the
+  synthetic load driver's, before Layout) plus a Layout entry for the three new files.
+- Not wired into `ci.yaml`, same as every other narrow scenario script
+  (`run_retention.sh`/`run_incident.sh`/`run_degraded.sh`/
+  `run_load_driver_shipper_test.sh`).

tools/e2e/README.md

@@ -252,39 +252,97 @@ accepted and acknowledged and that the driver's ledger names the exact same reco
 Vector actually decoded — not just a matching count. Not wired into `ci.yaml`, same as
 the retention/incident/degraded scenarios above.
 
-The saturation probe (`scripts/saturation_probe.py`, #377) and the ramp controller
-(`scripts/ramp_controller.py`, #379) are the other two pieces landed so far. Given a
-driver, a probe, and a `RampPolicy` of ascending load levels, `find_knee()` drives each
-level in turn and asks the probe whether it saturated, stopping at the first tripped
-level and reporting the knee: the highest level whose verdict stayed clear, plus the
-level that tripped it. A policy whose every level stays clear is reported as
-`BOUND_NOT_FOUND` rather than a fabricated ceiling at the top of the ramp — the failure
-mode #323's PRD calls out as mattering most. The controller touches no I/O and knows
-nothing about what a "level" means (a connection count, a rate, an upload
-concurrency, …); `driver` and `probe` are plain callables, so `test_ramp_controller.py`
-covers it entirely with in-process fakes, including the ramp-policy boundary cases
-(first step, last step, a single-step ramp).
-
-The curve reporter (`scripts/curve_reporter.py`, #380) turns a set of per-axis runs —
-each an ascending sequence of levels tried, whether each level was actually driven and
-measured or is a projected `EXTRAPOLATED` point, its real-vs-synthetic composition, and
-a CPU/memory/disk `ResourceSample` where one was taken — into a stable, diffable
-`CurveReport` (`build_report()`) plus a human-readable summary (`render_summary()`). It
-is a pure function, decoupled from `ramp_controller`/`saturation_probe` the same way
-those two are decoupled from each other: no I/O, no containers, its own `RunOutcome`
-mirroring `RampOutcome`'s values. `build_report()` sorts axes by name and each axis's
-points by level so the same set of runs always yields the same report regardless of
-collection order — required for a diff between two runs to show only a real change. The
-binding constraint at saturation (CPU, memory, or disk) is the resource that peaked
-highest across a run's *measured* points only (CPU-then-memory-then-disk as the
-deterministic tie-break), and is reported as unavailable rather than guessed when the
-axis never saturated or its knee has no accompanying resource sample. An extrapolated
-point never carries a resource sample in either the structured report or the summary —
-labelling every point `measured`/`extrapolated` in both is how the reporter avoids
-presenting a projection as a measurement, per #323's PRD.
+The saturation probe (`scripts/saturation_probe.py`, #377) is a third piece landed
+alongside it: a pure function over a window of ack-latency/unacked-window-depth/disk-
+buffer observations, checked in the PRD's stated priority order and unit-tested alone.
+
+The ramp controller (`scripts/ramp_controller.py`, #379) is the fourth. Given a driver, a
+probe, and a `RampPolicy` of ascending load levels, `find_knee()` drives each level in
+turn and asks the probe whether it saturated, stopping at the first tripped level and
+reporting the knee: the highest level whose verdict stayed clear, plus the level that
+tripped it. A policy whose every level stays clear is reported as `BOUND_NOT_FOUND`
+rather than a fabricated ceiling at the top of the ramp — the failure mode #323's PRD
+calls out as mattering most. The controller touches no I/O and knows nothing about what a
+"level" means (a connection count, a rate, an upload concurrency, …); `driver` and `probe`
+are plain callables, so `test_ramp_controller.py` covers it entirely with in-process
+fakes, including the ramp-policy boundary cases (first step, last step, a single-step
+ramp).
+
+The curve reporter (`scripts/curve_reporter.py`, #380) is the fifth. It turns a set of
+per-axis runs — each an ascending sequence of levels tried, whether each level was
+actually driven and measured or is a projected `EXTRAPOLATED` point, its real-vs-
+synthetic composition, and a CPU/memory/disk `ResourceSample` where one was taken — into
+a stable, diffable `CurveReport` (`build_report()`) plus a human-readable summary
+(`render_summary()`). It is a pure function, decoupled from `ramp_controller`/
+`saturation_probe` the same way those two are decoupled from each other: no I/O, no
+containers, its own `RunOutcome` mirroring `RampOutcome`'s values. `build_report()` sorts
+axes by name and each axis's points by level so the same set of runs always yields the
+same report regardless of collection order — required for a diff between two runs to
+show only a real change. The binding constraint at saturation (CPU, memory, or disk) is
+the resource that peaked highest across a run's *measured* points only (CPU-then-memory-
+then-disk as the deterministic tie-break), and is reported as unavailable rather than
+guessed when the axis never saturated or its knee has no accompanying resource sample. An
+extrapolated point never carries a resource sample in either the structured report or the
+summary — labelling every point `measured`/`extrapolated` in both is how the reporter
+avoids presenting a projection as a measurement, per #323's PRD.
 `tools/e2e/test/test_curve_reporter.py` covers all of this against synthetic run data
-alone. The topology composer #323 describes alongside these four modules is the one
-not-yet-implemented piece of that epic.
+alone.
+
+## Two-tier topology composer (#381)
+
+The sixth piece of #323's limits harness: brings up, with plain podman on a dedicated
+bridge network, the topology #323's PRD is ultimately sized around — a private site where
+robots have no direct path to the cloud, and an edge server aggregates their Records and
+forwards them upstream:
+
+```sh
+./tools/e2e/scripts/run_limits_two_tier.sh
+```
+
+A shared Postgres + RustFS (the real Destinations), an aggregating Shipper (a standalone
+Vector instance, `type = "vector"` + `type = "fluent"` sources, no dc_bridge/ROS
+involved), N real DC stacks (`params/e2e_limits_params.yaml`, each its own dc_bringup +
+dc_bridge + local Shipper), and M synthetic senders (one `scripts/load_driver.py`
+invocation, #378). At a fixed, non-saturating load, it reuses `verify_zero_loss.py`
+**unmodified** to prove Records survive the extra hop.
+
+Each real stack's own local Shipper has no Destination that writes to the real Postgres
+at all — `params/e2e_limits_forward_sink.toml` (an ADR-0003 `custom_config_files`
+passthrough, the same mechanism `e2e_passthrough_sink.toml`/`e2e_mcap_sink.toml` already
+use) wires a `type = "vector"` sink to the same `dc.<tag>` route branches a blessed
+Destination would, relaying them over the network to the aggregating Shipper via Vector's
+own native inter-instance protocol — the standard way to chain a local ("agent") Vector to
+a central ("aggregator") one. The aggregating Shipper's own `postgres` sink is the only
+thing that ever writes Measurement/synth Records into the real `dc_records` table. The M
+synthetic senders hit the aggregating Shipper's `fluent` source directly, adding the
+realistic connection count and byte rate #323's PRD calls for; their frames aren't
+`dc_records`-shaped, so they land on a separate `blackhole` sink and are verified via the
+driver's own sent/acked ledger instead of a Postgres row check.
+
+Two discoveries came out of implementing this, both recorded rather than worked around:
+
+1. **Files bypass the two-tier chain.** The Uploader — File bytes over its own AWS SDK
+   client, plus its own file-metadata Records — talks directly to whatever `s3`/
+   `postgres` Destination host is configured, with no Vector hop at all
+   (`dc_bridge/src/bridge_node.cpp`'s `run_uploader_worker` never touches the rendered
+   Vector config). There is no way to route Files through the aggregating Shipper without
+   new DC code, so `e2e_limits_params.yaml` points the Files-side Destinations straight at
+   the real shared Postgres/RustFS. Extending true two-tier coverage to Files is left as a
+   follow-up.
+2. **The aggregating Shipper's one `postgres` sink is a shared bottleneck N real stacks
+   don't have in any single-tier scenario** (each of those writes straight to its own
+   Postgres connection). Verified empirically: two real stacks reliably produced far more
+   at-least-once re-delivery than one does — past what `verify_zero_loss.py`'s
+   zero-tolerance memory-timestamp-uniqueness check accepts, even though every value still
+   arrived (the "duplicates" are identical `(date, value)` pairs, confirmed against
+   Postgres directly — genuine at-least-once repeats, not loss). `DC_E2E_LIMITS_REAL_STACKS`
+   stays configurable rather than hardcoded — finding exactly where that bottleneck sits is
+   the ramp controller's job (#323's next piece), not this composer's — but its default is
+   1 real stack so this script passes reliably on its own, matching #381's own scope
+   ("before any ramping logic exists, at a fixed, non-saturating load").
+
+Not wired into `ci.yaml`, same as the retention/incident/degraded/load-driver scenarios
+above.
 
 ## Layout
 
@@ -385,6 +443,10 @@ not-yet-implemented piece of that epic.
   structured report plus a human-readable summary naming the binding constraint (CPU,
   memory, or disk) at saturation, with every point labelled measured or extrapolated.
   No I/O, no container; unit-tested against synthetic run data alone.
+- `scripts/run_limits_two_tier.sh` / `params/e2e_limits_params.yaml` /
+  `params/e2e_limits_forward_sink.toml` — the two-tier topology composer (#381, part of
+  #323) described above: N real DC stacks + M synthetic senders through an aggregating
+  Shipper to shared Postgres/RustFS, reusing `verify_zero_loss.py` unmodified.
 
 ## `.dockerignore`
 

tools/e2e/params/e2e_limits_forward_sink.toml

@@ -0,0 +1,32 @@
+# SPDX-FileCopyrightText: 2022-2026 David Bensoussan
+# SPDX-License-Identifier: MPL-2.0
+
+# #381's "remote-Shipper-target" hop: an ADR-0003 `custom_config_files` passthrough sink
+# — a component dc_bridge has no knowledge of, same mechanism as
+# e2e_passthrough_sink.toml/e2e_mcap_sink.toml — wired to the *same* public `dc.<tag>`
+# routes e2e_limits_params.yaml's `local_sink` Destination creates. Where those two
+# existing snippets are `file`/`socket` sinks that stay inside the container, this one is
+# `type = "vector"`: Vector's own native relay protocol, the standard mechanism for
+# chaining a local ("agent") Vector to a central ("aggregator") one — verified directly
+# against the pinned Vector build before this file was written (a `fluent` source ->
+# `vector` sink -> `vector` source -> `file` sink chain, relaying load_driver.py frames
+# end to end with every custom field intact).
+#
+# `address` names run_limits_two_tier.sh's aggregating Shipper container on that script's
+# own bridge network (container-name DNS, same convention as
+# e2e_degraded_params.yaml's Postgres/RustFS hosts) — never 127.0.0.1: this Bridge's own
+# local Shipper's only path to the real Postgres is through that remote hop, standing in
+# for a robot with no direct route to the cloud (#323's motivating scenario).
+[sinks.to_aggregator]
+type = "vector"
+inputs = ["dc.dc.measurement.memory", "dc.dc.measurement.os", "dc.dc.measurement.storage", "dc.dc.measurement.uptime", "dc.dc.measurement.tcp_health", "dc.dc.measurement.dummy", "dc.dc.measurement.synth00", "dc.dc.measurement.synth01", "dc.dc.measurement.synth02", "dc.dc.measurement.synth03", "dc.dc.measurement.synth04", "dc.dc.measurement.synth05", "dc.dc.measurement.synth06", "dc.dc.measurement.synth07", "dc.dc.measurement.synth08", "dc.dc.measurement.synth09", "dc.dc.measurement.synth10", "dc.dc.measurement.synth11", "dc.dc.measurement.synth12", "dc.dc.measurement.synth13"]
+address = "dc-e2e-limits-agg:6000"
+
+# Same reasoning as the passthrough/MCAP snippets' own buffer block: a snippet's sink
+# gets Vector's small in-memory default (500 events) unless it asks for the same disk
+# buffer a blessed Destination gets — this is the sink actually carrying the
+# at-least-once guarantee across the extra hop, so it needs real durability under an
+# aggregator outage, not just the two examples that stay local.
+[sinks.to_aggregator.buffer]
+type = "disk"
+max_size = 268435488