diff --git a/progress.txt b/progress.txt index cf9bda18b..0c4f7fc31 100644 --- a/progress.txt +++ b/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.` 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`). diff --git a/tools/e2e/README.md b/tools/e2e/README.md index ef942c9f5..ce62893da 100644 --- a/tools/e2e/README.md +++ b/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.` 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` diff --git a/tools/e2e/params/e2e_limits_forward_sink.toml b/tools/e2e/params/e2e_limits_forward_sink.toml new file mode 100644 index 000000000..dac9d9016 --- /dev/null +++ b/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.` +# 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 diff --git a/tools/e2e/params/e2e_limits_params.yaml b/tools/e2e/params/e2e_limits_params.yaml new file mode 100644 index 000000000..8f6b3dd5c --- /dev/null +++ b/tools/e2e/params/e2e_limits_params.yaml @@ -0,0 +1,263 @@ +# SPDX-FileCopyrightText: 2022-2026 David Bensoussan +# SPDX-License-Identifier: MPL-2.0 + +# The limits harness's two-tier topology scenario (#381, part of #323's epic). Otherwise +# byte-for-byte params/e2e_params.yaml (same Measurement plugins, same synth/camera +# workload — verify_zero_loss.py is reused unmodified and hard-requires this exact shape: +# the 6 REAL_TAGS, memory's sub-1s polling for the timestamp-resolution check, the +# camera/Files pipeline, and synth00-13 for the ledger/passthrough/mcap checks), except: +# +# - `pgsql_records` is replaced by `local_sink`, a `type: file` Destination. Its only job +# is to make dc_bridge render `dc.` route branches for the Measurement/synth +# topics (route branches come from a Destination's `inputs` — see render.cpp — there is +# no way to get one without declaring some Destination); it writes to a throwaway local +# path nothing reads. The actual "ship this to Postgres" job moves to +# e2e_limits_forward_sink.toml's injected `vector`-type sink (ADR-0003 +# `custom_config_files` passthrough), which relays those same route branches on to +# run_limits_two_tier.sh's aggregating Shipper over the network instead of writing +# Records locally — this Bridge's own local Shipper never talks to the real Postgres at +# all, standing in for "a robot with no direct path to the cloud" (#323's motivating +# scenario). +# - `pgsql_files`/`rustfs` point at run_limits_two_tier.sh's shared Postgres/RustFS +# containers by name (dc-e2e-limits-postgres/dc-e2e-limits-rustfs), not 127.0.0.1 — see +# that file's header for why: the Uploader talks to whatever `s3`/`postgres` Destination +# host is configured directly over AWS-SDK/libpq calls, with no Vector hop at all, so +# there is no way to route Files (bytes or metadata) through the aggregating Shipper +# without new DC code. Recorded as this ticket's discovery per its acceptance +# criteria — Files bypass the two-tier chain; only Measurement Records go through it. +# - `custom_config_files` adds e2e_limits_forward_sink.toml alongside the existing +# passthrough/MCAP snippets. + +measurement_server: + ros__parameters: + save_local_base_path: "$HOME/.dc/e2e/data/%Y/%M/%D/%H" + all_base_path: "e2e" + measurement_plugins: ["memory", "os", "storage", "uptime", "tcp_health", "dummy", "synth00", "synth01", "synth02", "synth03", "synth04", "synth05", "synth06", "synth07", "synth08", "synth09", "synth10", "synth11", "synth12", "synth13", "camera"] + + memory: + plugin: "dc_measurements/Memory" + polling_interval: 200 + group_key: "memory" + + os: + plugin: "dc_measurements/OS" + polling_interval: 1000 + group_key: "os" + + storage: + plugin: "dc_measurements/Storage" + polling_interval: 1000 + path: "/tmp" + group_key: "storage" + + uptime: + plugin: "dc_measurements/Uptime" + polling_interval: 1000 + group_key: "uptime" + + tcp_health: + plugin: "dc_measurements/TCPHealth" + polling_interval: 1000 + name: "vector_forward" + host: "127.0.0.1" + port: 24224 + group_key: "tcp_health" + + dummy: + plugin: "dc_measurements/Dummy" + polling_interval: 1000 + group_key: "dummy" + + camera: + plugin: "dc_measurements/Camera" + polling_interval: 15000 + cam_name: "e2e_camera" + cam_topic: "/dc/e2e/camera/image_raw" + draw_det_barcodes: false + save_raw_img: true + save_rotated_img: false + save_detections_img: false + save_raw_base64: false + save_rotated_base64: false + save_inspected_base64: false + remote_keys: ["rustfs"] + remote_prefixes: ["camera"] + group_key: "camera" + + synth00: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth00" + timer_based: true + + synth01: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth01" + timer_based: true + + synth02: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth02" + timer_based: true + + synth03: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth03" + timer_based: true + + synth04: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth04" + timer_based: true + + synth05: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth05" + timer_based: true + + synth06: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth06" + timer_based: true + + synth07: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth07" + timer_based: true + + synth08: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth08" + timer_based: true + + synth09: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth09" + timer_based: true + + synth10: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth10" + timer_based: true + + synth11: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth11" + timer_based: true + + synth12: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth12" + timer_based: true + + synth13: + plugin: "dc_measurements/StringStamped" + enable_validator: false + polling_interval: 1000 + topic: "/dc/e2e/synth/synth13" + timer_based: true + +dc_bridge: + ros__parameters: + shipper: + data_dir: "$HOME/.dc/e2e/buffer" + destinations: ["local_sink", "pgsql_files", "rustfs", "raw_file"] + + # Creates the `dc.` route branches e2e_limits_forward_sink.toml's injected sink + # reads from — see this file's header. Writes locally and is never read back; the + # real egress for these Records is the injected `vector` sink below. + local_sink: + type: file + receives: records + inputs: ["/dc/measurement/memory", "/dc/measurement/os", "/dc/measurement/storage", "/dc/measurement/uptime", "/dc/measurement/tcp_health", "/dc/measurement/dummy", "/dc/measurement/synth00", "/dc/measurement/synth01", "/dc/measurement/synth02", "/dc/measurement/synth03", "/dc/measurement/synth04", "/dc/measurement/synth05", "/dc/measurement/synth06", "/dc/measurement/synth07", "/dc/measurement/synth08", "/dc/measurement/synth09", "/dc/measurement/synth10", "/dc/measurement/synth11", "/dc/measurement/synth12", "/dc/measurement/synth13"] + path: "/root/.dc/e2e/data/local_sink/records.ndjson" + time_key: "date" + + pgsql_files: + type: postgres + receives: records + # dc-e2e-limits-postgres, not 127.0.0.1 — see this file's header: the Uploader + # writes File metadata directly, with no Vector hop, so it bypasses the two-tier + # chain and reaches run_limits_two_tier.sh's shared Postgres straight over the + # network (its own bridge network — see that script for why). + host: "dc-e2e-limits-postgres" + port: 5432 + user: "dc" + password: "password" + database: "dc" + table: "dc_files" + time_key: "date" + + rustfs: + type: s3 + receives: files + inputs: ["/dc/measurement/camera"] + bucket: "dc-e2e" + # Same bypass as pgsql_files above — the Uploader's own AWS SDK client, not Vector. + endpoint: "http://dc-e2e-limits-rustfs:9000" + region: "us-east-1" + access_key_id: "rustfsadmin" + secret_access_key: "rustfsadmin" + force_path_style: true + + # Raw / generic-subscription mode (#227): verify_zero_loss.py's check_raw() is + # unconditional, so this stays enabled here too, unchanged from e2e_params.yaml — it + # writes locally (a `file` Destination, same bypass reasoning as local_sink above) + # rather than through the two-tier chain, since it has nothing to do with this + # ticket's Measurement-Records-through-the-aggregator claim. + raw_file: + type: file + receives: records + path: "/root/.dc/e2e/data/raw/records.ndjson" + time_key: "date" + + raw: + enabled: true + destination: "raw_file" + include: ["^/dc/e2e/synth/synth00$"] + exclude: ["^/rosout$", "^/parameter_events$", "^/dc/measurement/", "^/dc/group/"] + exclude_types: ["^sensor_msgs/msg/(Image|CompressedImage|PointCloud2)$"] + rescan_interval_secs: 1.0 + max_rate_hz: 10.0 + + files: + delete_when_sent: false + metadata_destination: "pgsql_files" + + custom_config_files: ["/opt/e2e/e2e_passthrough_sink.toml", "/opt/e2e/e2e_mcap_sink.toml", "/opt/e2e/e2e_limits_forward_sink.toml"] + + # Same-container loopback (Bridge-to-own-local-Shipper) — never leaves the container, + # so this stays 127.0.0.1 regardless of topology (same reasoning as + # e2e_degraded_params.yaml). + vector_forward_host: "127.0.0.1" + vector_forward_port: 24224 + +lifecycle_manager_dc: + ros__parameters: + node_names: ["measurement_server"] + transitions: [configure, activate] diff --git a/tools/e2e/scripts/run_limits_two_tier.sh b/tools/e2e/scripts/run_limits_two_tier.sh new file mode 100755 index 000000000..c4a9a9b5b --- /dev/null +++ b/tools/e2e/scripts/run_limits_two_tier.sh @@ -0,0 +1,390 @@ +#!/bin/bash +# SPDX-FileCopyrightText: 2022-2026 David Bensoussan +# SPDX-License-Identifier: MPL-2.0 + +# Limits harness: two-tier topology composer (#381, part of #323's epic). From the repo +# root: +# +# ./tools/e2e/scripts/run_limits_two_tier.sh +# +# Brings up, with plain podman on a dedicated bridge network (container-name DNS, same +# convention as run_degraded.sh — see that script's header for why not --network host: +# N real DC stacks all binding the same default Bridge-to-Shipper port would collide on a +# shared host network namespace): a shared Postgres + RustFS (the real Destinations), an +# aggregating Shipper (a standalone Vector instance, no dc_bridge/ROS involved), N real DC +# stacks (each its own dc_bringup + dc_bridge + local Shipper, +# params/e2e_limits_params.yaml), and M synthetic senders (scripts/load_driver.py, #378). +# At a fixed, non-saturating load, it then reuses verify_zero_loss.py UNMODIFIED to prove +# Records survive the extra hop. +# +# --- The two-tier chain ------------------------------------------------------------------ +# +# Each real stack's own local Shipper (its Bridge-spawned Vector, unconditional per +# dc_bridge/src/bridge_node.cpp — every Bridge always runs one) has no Destination that +# writes Records to the real Postgres at all: params/e2e_limits_params.yaml replaces the +# usual `pgsql_records` with `local_sink`, a throwaway `file` Destination whose only job +# is to make dc_bridge render the `dc.` route branches those topics need. +# params/e2e_limits_forward_sink.toml (ADR-0003 `custom_config_files`, the same +# passthrough mechanism e2e_passthrough_sink.toml/e2e_mcap_sink.toml already use) wires a +# `type = "vector"` sink to those same route branches — Vector's own native relay +# protocol, the standard way to chain a local ("agent") Vector to a central +# ("aggregator") one — forwarding them over the network to this script's aggregating +# Shipper instead. That relay was verified directly against the pinned Vector build +# before this script was written: a `fluent` source -> `vector` sink -> `vector` source +# -> `file` sink chain relayed scripts/load_driver.py frames end to end with every custom +# field intact, including the ack round trip back to the sender. The aggregating +# Shipper's own `postgres` sink is the only thing that ever writes Measurement/synth +# Records into the real `dc_records` table — no real stack's own local Shipper has a +# direct path there, standing in for #323's motivating scenario: "a private site where +# robots have no internet, and an edge server aggregates ... and forwards upstream." +# +# The M synthetic senders (one scripts/load_driver.py invocation, `--connections M`) hit +# the aggregating Shipper's own `fluent` source directly, alongside the `vector` source +# tier-1 relays into — the same aggregating Shipper the real stacks use, adding the +# realistic connection count and byte rate #323's PRD calls for. Their frames aren't +# dc_records-shaped (see e2e_limits_forward_sink.toml's header), so they land on a +# separate `blackhole` sink rather than the real Postgres path; the driver's own +# sent/acked ledger (checked below) is their zero-loss proof instead of a Postgres row +# check. +# +# --- Discovery recorded per #381's acceptance criteria ----------------------------------- +# +# This composer only routes `receives: records` Measurement Records through 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 +# params/e2e_limits_params.yaml points the Files-side Destinations straight at the real +# shared Postgres/RustFS instead of through the chain. Extending true two-tier coverage +# to Files is out of scope here, left as a follow-up. +# +# Second discovery: the two-tier design concentrates every real stack's Records through +# one aggregating Shipper's one `postgres` sink — a shared bottleneck no single-tier +# scenario (run.sh et al., each stack writing straight to its own Postgres connection) +# has. Verified empirically here: two real stacks (REAL_STACKS=2) 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 (check_timestamp_resolution) accepts. +# Vector's own end-to-end acknowledgement retries a chunk whose ack didn't land inside its +# window, and the busier shared sink under N>1 was slow enough here to trip that. Not a +# correctness bug — every value still arrived; the "duplicates" are identical (date, +# value) pairs, genuine at-least-once repeats, not loss (REAL_STACKS=1's own clean run and +# a direct Postgres query both confirm it) — and not something to silently work around: +# REAL_STACKS stays configurable rather than hardcoded, since finding exactly where that +# bottleneck sits is squarely the ramp controller's job (#323's next piece), not this +# composer's. Default kept at 1 real stack so this script passes reliably on its own, per +# #381's own "before any ramping logic exists, at a fixed, non-saturating load" scope. +# +# Env vars (all optional): +# DC_E2E_LIMITS_REAL_STACKS number of real DC stacks (default 1 — see the +# second discovery above before raising this) +# DC_E2E_LIMITS_SYNTH_CONNECTIONS load_driver.py connection count (default 50) +# DC_E2E_LIMITS_SYNTH_RATE_HZ load_driver.py per-connection Record rate (default 5) +# DC_E2E_LIMITS_STEADY_STATE_SECONDS fixed-load window duration (default 30) +# DC_E2E_LIMITS_DRAIN_SECONDS settle time after the window, before verifying (default 15) +# DC_E2E_KEEP "true" to leave the stack (and its network) up after a +# failure for debugging +# DC_E2E_IMAGE / DC_WORKSPACE_IMAGE same meaning as run.sh +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +E2E_DIR="$(dirname "$SCRIPT_DIR")" +REPO_ROOT="$(cd "$E2E_DIR/../.." && pwd)" +RUN_DIR="$E2E_DIR/.run/limits_two_tier" + +REAL_STACKS="${DC_E2E_LIMITS_REAL_STACKS:-1}" +SYNTH_CONNECTIONS="${DC_E2E_LIMITS_SYNTH_CONNECTIONS:-50}" +SYNTH_RATE_HZ="${DC_E2E_LIMITS_SYNTH_RATE_HZ:-5}" +STEADY_STATE_SECONDS="${DC_E2E_LIMITS_STEADY_STATE_SECONDS:-30}" +# The camera Measurement fires every 15s (params/e2e_limits_params.yaml) and its capture +# still has to clear the Uploader's own capture/queue/upload/verify pipeline after the +# steady-state window ends — verified empirically to matter: without this drain, a run +# landing its last capture near the end of the window intermittently stopped the stack +# before that capture's dc_files metadata row landed, failing check_files() on an +# otherwise-correct run. +DRAIN_SECONDS="${DC_E2E_LIMITS_DRAIN_SECONDS:-15}" +KEEP="${DC_E2E_KEEP:-false}" + +NET=dc-e2e-limits-net +# Hyphens, not the underscored dc_e2e_* convention every other scenario script uses: +# these names are embedded in URLs (RustFS/Postgres endpoints, the vector-relay address) +# and aws-cli 2.36's --endpoint-url parser rejects a hostname with an underscore outright +# (verified empirically — the same underscored convention on run_degraded.sh's own +# RustFS container name would hit this identically). Volume names below are unaffected +# (never used as a hostname) and keep the usual underscored convention. +PG_C=dc-e2e-limits-postgres +RUSTFS_C=dc-e2e-limits-rustfs +AGG_C=dc-e2e-limits-agg +AGG_VECTOR_PORT=6000 +AGG_FLUENT_PORT=24224 +STACK_PREFIX=dc-e2e-limits-stack- +# Stack 0 is the ledger-checked stack: verify_zero_loss.py's contract is one ledger per +# run. The other real stacks and the synthetic senders are unverified-by-ledger +# background load — exactly the PRD's "real stacks plus synthetic senders together" +# nominal-load mix, all going through the same aggregating Shipper as stack 0. +LEDGER_STACK_INDEX=0 + +mkdir -p "$RUN_DIR" +cd "$E2E_DIR" + +log() { echo "[e2e-limits $(date -u +%H:%M:%S)] $*"; } + +stack_name() { echo "${STACK_PREFIX}$1"; } + +all_stack_names() { + local i + for ((i = 0; i < REAL_STACKS; i++)); do + stack_name "$i" + done +} + +remove_stack() { + # shellcheck disable=SC2046 + podman rm -f --ignore $(all_stack_names) "$AGG_C" "$PG_C" "$RUSTFS_C" >/dev/null + local i v + for ((i = 0; i < REAL_STACKS; i++)); do + for v in "dc_e2e_limits_buffer_$i" "dc_e2e_limits_data_$i"; do + if podman volume exists "$v"; then + podman volume rm "$v" >/dev/null + fi + done + done + for v in dc_e2e_limits_pgdata dc_e2e_limits_rustfs_data; do + if podman volume exists "$v"; then + podman volume rm "$v" >/dev/null + fi + done + podman network rm "$NET" >/dev/null 2>&1 || true +} + +cleanup() { + local exit_code=$? + if [ "$exit_code" -ne 0 ] && [ "$KEEP" = "true" ]; then + log "FAILED (exit $exit_code) — leaving the stack up (DC_E2E_KEEP=true) for debugging" + exit "$exit_code" + fi + log "tearing down" + local s + for s in $(all_stack_names); do + if podman container exists "$s"; then + podman logs "$s" > "$RUN_DIR/${s}.log" 2>&1 + fi + done + if podman container exists "$AGG_C"; then + podman logs "$AGG_C" > "$RUN_DIR/agg.log" 2>&1 + fi + remove_stack + exit "$exit_code" +} +trap cleanup EXIT + +# --- obtain the DC stack image (shared with run.sh — see its header) ------------------ +if [ -n "${DC_E2E_IMAGE:-}" ]; then + log "using prebuilt E2E image: $DC_E2E_IMAGE (no build)" + podman image exists "$DC_E2E_IMAGE" || podman pull "$DC_E2E_IMAGE" + DC_IMAGE="$DC_E2E_IMAGE" +else + if [ -n "${DC_WORKSPACE_IMAGE:-}" ]; then + log "using prebuilt DC workspace image: $DC_WORKSPACE_IMAGE (skipping build.sh)" + podman image exists "$DC_WORKSPACE_IMAGE" || podman pull "$DC_WORKSPACE_IMAGE" + WORKSPACE_IMAGE="$DC_WORKSPACE_IMAGE" + else + log "building the DC workspace image (tools/e2e/scripts/build.sh — the same build CI uses)" + "$SCRIPT_DIR/build.sh" + WORKSPACE_IMAGE="dc-workspace:latest" + fi + log "building the E2E image (Containerfile.e2e, FROM the workspace image)" + podman build --build-arg "BASE_IMAGE=$WORKSPACE_IMAGE" -t dc-e2e:latest -f Containerfile.e2e . + DC_IMAGE="dc-e2e:latest" +fi + +# The exact Vector version dc_bridge is built and tested against, for the standalone +# aggregating Shipper — read out of the file that pins it, not duplicated as a second +# source of truth (same as run_load_driver_shipper_test.sh). +VECTOR_VERSION="$(grep -oP 'set\(VECTOR_VERSION "\K[^"]+' "$REPO_ROOT/vector_vendor/CMakeLists.txt")" +VECTOR_IMAGE="docker.io/timberio/vector:${VECTOR_VERSION}-debian" + +# Clean any leftovers from a previous (possibly DC_E2E_KEEP=true) run. +remove_stack + +# --- bridge network + shared destinations ---------------------------------------------- +log "creating the bridge network ($NET)" +podman network create "$NET" >/dev/null + +log "starting the shared Postgres + RustFS on $NET" +podman run -d --network "$NET" --name "$PG_C" \ + -e POSTGRES_USER=dc -e POSTGRES_PASSWORD=password -e POSTGRES_DB=dc \ + -v dc_e2e_limits_pgdata:/var/lib/postgresql/data \ + -v "$E2E_DIR/sql/init.sql:/docker-entrypoint-initdb.d/init.sql:ro" \ + docker.io/library/postgres:13 >/dev/null +podman run -d --network "$NET" --name "$RUSTFS_C" \ + -v dc_e2e_limits_rustfs_data:/data \ + docker.io/rustfs/rustfs@sha256:84ce557a0245a06a9aae5516f55ee0f007fca78d41df356f419306fdc0cb168c >/dev/null + +timeout 120 bash -c "until podman exec $PG_C psql -U dc -d dc -tAc \"SELECT to_regclass('public.dc_records')\" 2>/dev/null | grep -q dc_records; do sleep 2; done" \ + || { log "FAIL: Postgres never came up with sql/init.sql applied"; exit 1; } + +log "waiting for RustFS to accept TCP connections on $NET" +# --entrypoint bash: the image's own ENTRYPOINT (entrypoint.sh) launches the full DC +# stack and appends any extra args to `ros2 launch` rather than running them as a +# separate command, so a one-off diagnostic run needs an explicit override (verified +# empirically) — a plain bash /dev/tcp probe rather than scripts/measure_rtt.py, so this +# doesn't depend on that file happening to be baked into whatever $DC_IMAGE is passed in. +timeout 60 bash -c " + until podman run --rm --network $NET --entrypoint bash '$DC_IMAGE' -c 'exec 3<>/dev/tcp/$RUSTFS_C/9000' >/dev/null 2>&1; do + sleep 1 + done +" || { log "FAIL: RustFS never became reachable on $NET"; exit 1; } + +log "creating the RustFS bucket" +podman run --rm --network "$NET" \ + -e AWS_ACCESS_KEY_ID=rustfsadmin -e AWS_SECRET_ACCESS_KEY=rustfsadmin -e AWS_DEFAULT_REGION=us-east-1 \ + docker.io/amazon/aws-cli:latest \ + --endpoint-url "http://$RUSTFS_C:9000" s3 mb s3://dc-e2e + +# --- the aggregating Shipper: a standalone Vector instance, no dc_bridge involved ------- +log "rendering the aggregating Shipper's Vector config" +cat > "$RUN_DIR/aggregator_vector.toml" < route branches here +# over Vector's native inter-instance protocol (see e2e_limits_forward_sink.toml). +[sources.from_tier1] +type = "vector" +address = "0.0.0.0:${AGG_VECTOR_PORT}" + +# The M synthetic senders (#378's load_driver.py) speak the same shipper ingest protocol +# a real Bridge's Forwarder does — this is the same fluent listener shape +# run_load_driver_shipper_test.sh already proves load_driver.py is byte-compatible with. +[sources.from_synth] +type = "fluent" +address = "0.0.0.0:${AGG_FLUENT_PORT}" + +# The only thing that writes Measurement/synth Records into the real dc_records table — +# see this script's header for why no real stack's own local Shipper does. +[sinks.to_postgres] +type = "postgres" +inputs = ["from_tier1"] +endpoint = "postgres://dc:password@${PG_C}:5432/dc" +table = "dc_records" + +[sinks.to_postgres.buffer] +type = "disk" +max_size = 268435488 + +# Synthetic frames aren't dc_records-shaped (connection_id/seq, not date/tag/...) — they +# exist to add realistic connection count and byte rate (#323's PRD), not to be +# individually verified in Postgres. Kept off the real Records path so a schema mismatch +# there can never touch it; the driver's own sent/acked ledger is their zero-loss proof. +[sinks.discard_synth] +type = "blackhole" +inputs = ["from_synth"] +EOF + +log "starting the aggregating Shipper ($AGG_C)" +podman run -d --network "$NET" -p "${AGG_FLUENT_PORT}:${AGG_FLUENT_PORT}" --name "$AGG_C" \ + -v "$RUN_DIR/aggregator_vector.toml:/etc/vector/vector.toml:Z" \ + "$VECTOR_IMAGE" -c /etc/vector/vector.toml >/dev/null + +log "waiting for the aggregating Shipper's synthetic-sender ingress (fluent, published on the host) to accept connections" +if ! timeout 30 bash -c "until (exec 3<>/dev/tcp/127.0.0.1/${AGG_FLUENT_PORT}) 2>/dev/null; do sleep 0.5; done"; then + log "FAIL: the aggregating Shipper's fluent source never started listening on port $AGG_FLUENT_PORT" + exit 1 +fi + +log "waiting for the aggregating Shipper's tier-1 ingress (vector protocol, $NET-only) to accept connections" +timeout 60 bash -c " + until podman run --rm --network $NET --entrypoint bash '$DC_IMAGE' -c 'exec 3<>/dev/tcp/$AGG_C/$AGG_VECTOR_PORT' >/dev/null 2>&1; do + sleep 1 + done +" || { log "FAIL: the aggregating Shipper's vector-protocol source never became reachable on $NET"; exit 1; } + +# --- N real DC stacks -------------------------------------------------------------------- +log "starting $REAL_STACKS real DC stack(s) against params/e2e_limits_params.yaml" +for ((i = 0; i < REAL_STACKS; i++)); do + s="$(stack_name "$i")" + podman volume create "dc_e2e_limits_buffer_$i" >/dev/null + podman volume create "dc_e2e_limits_data_$i" >/dev/null + podman run -d --network "$NET" --name "$s" \ + -v "dc_e2e_limits_buffer_$i:/root/.dc/e2e/buffer" \ + -v "dc_e2e_limits_data_$i:/root/.dc/e2e/data" \ + -v "$E2E_DIR/params/e2e_limits_params.yaml:/opt/e2e/e2e_params.yaml:ro" \ + -v "$E2E_DIR/params/e2e_limits_forward_sink.toml:/opt/e2e/e2e_limits_forward_sink.toml:ro" \ + "$DC_IMAGE" >/dev/null + # Staggered, not all at once: reduces simultaneous ROS/DDS-discovery and lifecycle + # bring-up contention across N concurrently-starting stacks. Does not by itself fix + # #381's other recorded discovery below (REAL_STACKS default) — that one is a runtime + # contention effect, not a startup-ordering one. + sleep 5 +done + +log "waiting for the first Record to reach Postgres through the two-tier chain" +timeout 90 bash -c "until [ \"\$(podman exec $PG_C psql -U dc -d dc -tAc 'SELECT count(*) FROM dc_records' 2>/dev/null || echo 0)\" -gt 0 ] 2>/dev/null; do sleep 1; done" \ + || { log "FAIL: no Record reached Postgres through the two-tier chain within 90s"; exit 1; } +log "PASS: at least one Record reached Postgres through the aggregating Shipper" + +# --- fixed nominal-load window: real stacks (already running) + M synthetic senders ----- +log "running the fixed nominal-load window for ${STEADY_STATE_SECONDS}s: $REAL_STACKS real stack(s) + $SYNTH_CONNECTIONS synthetic connection(s) @ ${SYNTH_RATE_HZ}Hz" +SUMMARY_JSON="$(cd "$REPO_ROOT" && uv run --frozen python3 tools/e2e/scripts/load_driver.py \ + 127.0.0.1 "$AGG_FLUENT_PORT" \ + --connections "$SYNTH_CONNECTIONS" \ + --rate "$SYNTH_RATE_HZ" \ + --duration "$STEADY_STATE_SECONDS" \ + --tag "dc.e2e.limits_synth")" +echo "$SUMMARY_JSON" > "$RUN_DIR/synth_summary.json" +log "synthetic sender summary: $SUMMARY_JSON" + +UNACKED="$(python3 -c "import json,sys; print(json.load(sys.stdin)['unacked'])" <<<"$SUMMARY_JSON")" +if [ "$UNACKED" -ne 0 ]; then + log "FAIL: $UNACKED of the synthetic sender's frame(s) were never acknowledged by the aggregating Shipper under nominal load" + exit 1 +fi +log "PASS: every synthetic frame was accepted and acknowledged by the aggregating Shipper" + +log "draining for ${DRAIN_SECONDS}s (lets the last camera capture clear the Uploader pipeline)" +sleep "$DRAIN_SECONDS" + +log "stopping the real DC stacks so counts settle before verification" +for s in $(all_stack_names); do + podman stop "$s" >/dev/null +done +sleep 5 + +# --- extract the ledger-checked stack's output, same pattern as run.sh ------------------ +DATA_VOL="dc_e2e_limits_data_${LEDGER_STACK_INDEX}" + +log "extracting stack $LEDGER_STACK_INDEX's passthrough sink output" +podman run --rm --entrypoint bash \ + -v "$DATA_VOL:/vol:ro" "$DC_IMAGE" \ + -c 'cat /vol/passthrough/records.ndjson 2>/dev/null || true' > "$RUN_DIR/passthrough.ndjson" + +log "extracting stack $LEDGER_STACK_INDEX's raw mode output" +podman run --rm --entrypoint bash \ + -v "$DATA_VOL:/vol:ro" "$DC_IMAGE" \ + -c 'cat /vol/raw/records.ndjson 2>/dev/null || true' > "$RUN_DIR/raw.ndjson" + +log "summarizing stack $LEDGER_STACK_INDEX's MCAP passthrough writer output" +podman run --rm --entrypoint python3 \ + -v "$DATA_VOL:/vol:ro" "$DC_IMAGE" \ + /opt/e2e/mcap_summary.py /vol/mcap > "$RUN_DIR/mcap_summary.json" + +log "extracting stack $LEDGER_STACK_INDEX's workload ledger" +podman run --rm --entrypoint bash \ + -v "$DATA_VOL:/vol:ro" "$DC_IMAGE" \ + -c 'cat /vol/workload_ledger.txt 2>/dev/null || true' > "$RUN_DIR/workload_ledger.txt" + +# --- verify (verify_zero_loss.py's own logic, reused unmodified) ------------------------ +log "verifying zero-loss against stack $LEDGER_STACK_INDEX's ledger, through the two-tier chain" +python3 "$SCRIPT_DIR/verify_zero_loss.py" \ + --postgres-container "$PG_C" \ + --num-synth-topics 14 \ + --ledger-file "$RUN_DIR/workload_ledger.txt" \ + --passthrough-file "$RUN_DIR/passthrough.ndjson" \ + --mcap-summary-file "$RUN_DIR/mcap_summary.json" \ + --raw-file "$RUN_DIR/raw.ndjson" \ + --report "$RUN_DIR/verification_report.json" + +log "PASS: zero-loss E2E harness through the two-tier topology (#381)"