test(dc_bridge,e2e): prove the pipeline under degraded network conditions (#366)

Extends the Forwarder's mock ingest peer with a test-only acknowledgement policy
(delay, never-ack probability, throttled drain) covering unacked window depth,
backpressure classification, and resend-under-delay at the existing unit-test seam
-- no container needed. Adds a sibling E2E scenario, run_degraded.sh, that shapes
the path to Postgres/RustFS with tc netem (via a netns-joining helper container,
verified to work unprivileged under rootless podman) behind a mandatory shaping
pre-check, induces a link-goes-away-without-the-process-dying fault distinct from
the existing outage scenario, and reuses verify_zero_loss.py unmodified (only an
additive --conditions-file flag). Named network profiles live in one declarative
place (network_profiles.py) shared by the scenario and, later, #323's harness.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TryRM4gK1GQE8uCDE4iPhX
Signed-off-by: David Bensoussan <d.bensoussan@proton.me>

Author: Minipada

dc_bridge/test/forwarder_test.cpp

@@ -7691,3 +7691,82 @@ the one-way `qrcodes_waypoint_follower.py` patrol never actually returns to the
 above is demonstrated by manual driving, not the autonomous demo — the issue's acceptance
 criteria ask for the capability and documented rates, not a scripted docking sequence nav2's own
 `docking_server` config still calls a "never docks" placeholder for.
+
+## #366 - Prove the pipeline under degraded network conditions: delayed acks and a shaped Destination link
+
+Every DC reliability claim to date was proven over intra-container loopback: microsecond RTT,
+zero loss, unbounded bandwidth, and a fault model where a Destination is either up or a stopped
+process. Real robots reach their Destinations over WiFi and a site uplink. This closes that gap
+at the two seams the PRD identified, without changing anything about the pipeline being measured.
+
+**Forwarder acknowledgement-policy mock peer (`dc_bridge/test/forwarder_test.cpp`).** The
+existing mock shipper ingest peer gains a test-only `AckPolicy` (ack delay, a probability of
+never acknowledging, a throttled drain rate) and a `run_ack_policy_peer()` driver built on a real
+`msgpack::unpacker` — same parsing discipline as Forwarder's own `drain_acks()`, so a drain rate
+that splits one frame across several `recv()` calls still parses correctly. Five new gtest cases:
+unacked window depth stays small (not the shipped 10000-record bound) at a realistic 100 Hz/50ms
+rate-delay combination; the same bound is exceeded by rate-vs-delay alone (not just a peer that
+never acks at all) and still converges once the burst ends; a slow-draining peer is classified as
+Backpressure with the connection retained, distinguishing it from a peer that's gone (Io, drops
+the connection); a resend past `ack_timeout` duplicates the chunk on the wire and the window still
+clears once the delayed original ack lands; an out-of-order ack evicts exactly the record it names
+(pinned via a 22-char placeholder chunk id — `generate_chunk_id()` base64-encodes 16 bytes without
+padding, always that length regardless of content).
+
+Two real bugs surfaced building this, both in the *test* harness, not Forwarder: the initial
+`run_ack_policy_peer` used a blocking `recv()` with no timeout, so once a test's client went idle
+the peer thread never re-checked its `stop` flag and `srv.join()` hung forever (fixed with a
+100ms `SO_RCVTIMEO`); and three tests used `ASSERT_*` ahead of `stop.store(true); srv.join()`,
+so a failing assertion returned early without joining a still-running `std::thread`, which
+`std::terminate()`s the whole test binary on destruction — fixed by demoting those to `EXPECT_*`
+with guarded follow-on checks, so one failing assertion reports as one failing test instead of
+aborting the entire suite. Built and run 5x against a scoped rebuild (dc_common + dc_bridge
+mounted over `localhost/dc-workspace:latest`, since a full `colcon build` wasn't warranted for a
+test-only change) — 17/17 passing, no flakiness, after `prek`'s `clang-format` reformatted the new
+code.
+
+**Network profile definitions (`tools/e2e/scripts/network_profiles.py`).** One declarative place
+for named, fixed conditions — `loopback` (unshaped), `good-wifi`, `poor-wifi`, `site-uplink` — each
+a `delay_ms`/`jitter_ms`/`loss_pct`/optional `rate_kbit`, with a CLI (`--shell` for a bash `eval`,
+`--tc-args` for a `tc qdisc ... netem` argument list, `--json`). An unknown name is a `ValueError`
+naming the known profiles, never a silent default. 10 pytest cases
+(`tools/e2e/test/test_network_profiles.py`).
+
+**`tools/e2e/scripts/run_degraded.sh` — the E2E sibling scenario.** Reuses `run.sh`'s image,
+reference workload shape, and `verify_zero_loss.py` unmodified, but Postgres/RustFS run on their
+own bridge network (`dc_e2e_deg_net`) instead of `--network host`, since nothing can be shaped
+without shaping the host itself under `--network host`. `params/e2e_degraded_params.yaml` is
+byte-for-byte `e2e_params.yaml` except the two Destination hosts (container-name DNS instead of
+`127.0.0.1`) — the Bridge-to-Shipper socket stays same-container loopback and out of scope, per
+the PRD.
+
+Traffic shaping was the PRD's stated single largest technical risk and was established first,
+empirically, before anything was built on it: rootless podman + `--cap-add=NET_ADMIN` on a
+short-lived helper container that joins a target's network namespace (`--network
+container:<name>`) can add a `tc netem` qdisc to that target's `eth0` with **no capability
+required on the target container itself** — verified with disposable alpine containers (a peer's
+measured RTT rose from <1ms to ~128ms after `netem delay 100ms 20ms loss 5%` was applied to the
+target's own interface from the joining helper). `run_degraded.sh` applies this via a
+`SHAPER_IMAGE` (alpine + iproute2, built once and cached) and `set_qdisc`/`clear_qdisc` helpers
+that track per-container qdisc state to choose `tc qdisc add` vs `change`. A shaping pre-check
+(`scripts/measure_rtt.py` — stdlib TCP connect-time sampling, since the Postgres/RustFS images
+ship neither `ping` nor `iputils`) measures the actual round trip before any workload runs and
+hard-fails the run if it isn't consistent with the requested profile, so a degraded run can never
+silently pass as loopback. A LINK fault (`tc netem loss 100%` on both Destinations, then restored)
+is induced partway through the run — the network disappearing without either process dying, a
+fault type distinct from `run.sh`'s container stop/restart, which resets Forwarder state and so
+tests something else. Conditions (profile + measured pre-check RTT) are written to
+`tools/e2e/.run/conditions.json`; `verify_zero_loss.py` gained one additive `--conditions-file`
+flag that folds that file into its report unchanged — no existing verification logic touched.
+
+Not wired into `ci.yaml`, matching the existing precedent for `run_retention.sh`/`run_incident.sh`
+(neither is CI-gating either) and the PRD's own "Out of Scope": whether this gates merges is left
+for once its runtime and stability are measured. The full container-network E2E path (bridge
+network + shaped Postgres/RustFS + the real DC stack under a degraded profile) was not run
+end-to-end here — the shaping *mechanism* was verified empirically in isolation (see above) and
+the script was built directly on that verified mechanism, but a full `run_degraded.sh` execution
+needs a full workspace image build (tens of minutes) that wasn't run as part of this session; that
+remains for a first real run to confirm.
+
+`tools/e2e/README.md` documents the new scenario and its layout entries; `Containerfile.e2e` gained
+`COPY` lines for `measure_rtt.py` (already needed by `run_degraded.sh`'s readiness/pre-check steps).

progress.txt

@@ -19,6 +19,9 @@ COPY params/e2e_params.yaml /opt/e2e/e2e_params.yaml
 COPY params/e2e_passthrough_sink.toml /opt/e2e/e2e_passthrough_sink.toml
 COPY params/e2e_mcap_sink.toml /opt/e2e/e2e_mcap_sink.toml
 COPY scripts/mcap_summary.py /opt/e2e/mcap_summary.py
+# run_degraded.sh (#366): stdlib-only, run as one-off containers on its shaped bridge
+# network for readiness polling and the shaping pre-check (see measure_rtt.py's header).
+COPY scripts/measure_rtt.py /opt/e2e/measure_rtt.py
 COPY scripts/entrypoint.sh /opt/e2e/entrypoint.sh
 RUN chmod +x /opt/e2e/entrypoint.sh
 

tools/e2e/Containerfile.e2e

@@ -182,6 +182,43 @@ broadcast node: that topic is the contract Measurements subscribe to, `dc_trigge
 its own tests for minting the event, and the broadcast node is not in `dc_bringup`'s
 launch yet. Not run by `ci.yaml`, same as the retention scenario.
 
+## Degraded-network scenario (#366)
+
+A fourth scenario, proving the zero-loss guarantee above holds under stated network
+conditions instead of only over intra-container loopback:
+
+```sh
+./tools/e2e/scripts/run_degraded.sh                         # loopback profile (baseline)
+DC_E2E_PROFILE=poor-wifi ./tools/e2e/scripts/run_degraded.sh
+```
+
+Same `dc-e2e` image and reference workload as `run.sh` (`params/e2e_degraded_params.yaml`
+is byte-for-byte `params/e2e_params.yaml` except for the two Destination hosts — see that
+file's header), but Postgres and RustFS run on their own bridge network instead of
+`--network host`, so the path to them can be shaped without shaping the host itself. A
+named profile (`scripts/network_profiles.py`: `loopback`, `good-wifi`, `poor-wifi`,
+`site-uplink` — one declarative place, so two runs of the same name are comparable) is
+applied as a `tc netem` qdisc on Postgres's and RustFS's own interfaces, via a short-lived
+helper container that joins their network namespace (`--network container:<name>
+--cap-add=NET_ADMIN`) rather than granting either image any capability itself. A shaping
+pre-check measures the actual TCP connect-time round trip before any workload runs
+(`scripts/measure_rtt.py`) and hard-fails the run if the measurement isn't consistent with
+the requested profile — a degraded run must never pass silently as loopback because
+shaping failed to apply. Partway through the run, a LINK fault (`tc netem loss 100%` on
+both Destinations, then restored) proves the network going away is handled distinctly from
+`run.sh`'s container stop/restart — nothing here resets Forwarder state. The network
+conditions a run used, plus the measured pre-check round trip, are written to
+`tools/e2e/.run/conditions.json` and folded into `verify_zero_loss.py`'s report — a result
+is never separable from the conditions that produced it.
+
+Not wired into `ci.yaml`, same as the retention and incident scenarios below: whether it
+gates merges is left for once its runtime and stability are measured (#366's own "Out of
+Scope"). The Forwarder's own unacked-window-depth, backpressure-classification, and
+resend-under-delay behaviour under acknowledgement delay is covered separately and much
+more cheaply at `dc_bridge/test/forwarder_test.cpp`'s existing mock-ingest-peer seam (an
+`AckPolicy` describing when/whether a chunk gets acknowledged) — no container needed for
+that; see that file for the corresponding unit tests.
+
 ## Layout
 
 - `Containerfile` — builds the full DC workspace (every `dc_*` package, all C++ since
@@ -260,6 +297,14 @@ launch yet. Not run by `ci.yaml`, same as the retention scenario.
   `dc-e2e` image.
 - `params/e2e_incident_params.yaml` / `scripts/run_incident.sh` — the incident-capture
   scenario (#291) described above, likewise reusing the same `dc-e2e` image.
+- `params/e2e_degraded_params.yaml` / `scripts/run_degraded.sh` — the degraded-network
+  scenario (#366) described above; a fourth sibling harness, reusing the same `dc-e2e`
+  image and `verify_zero_loss.py`, but its own bridge network so the path to the
+  Destinations can be shaped.
+- `scripts/network_profiles.py` — the named network conditions (#366) `run_degraded.sh`
+  applies, in one declarative place also meant for #323's limits harness to reuse.
+- `scripts/measure_rtt.py` — stdlib TCP connect-time sampler (#366), used both for
+  Destination readiness polling and the shaping pre-check `run_degraded.sh` never skips.
 
 ## `.dockerignore`
 

tools/e2e/README.md

@@ -0,0 +1,237 @@
+# SPDX-FileCopyrightText: 2022-2026 David Bensoussan
+# SPDX-License-Identifier: MPL-2.0
+
+# The degraded-network scenario's (#366) params: byte-for-byte params/e2e_params.yaml
+# except for the two Destination hosts, which name containers on run_degraded.sh's own
+# bridge network (dc_e2e_deg_net) instead of 127.0.0.1 — the shaped participants can't
+# share run.sh's --network host, or nothing could be shaped without shaping the host
+# itself (see run_degraded.sh's header). vector_forward_host and tcp_health.host stay
+# 127.0.0.1: that socket is the Bridge-to-Shipper path, which never leaves the
+# container's own network namespace (the Bridge forks and supervises the Shipper as a
+# child process) and is out of scope for this PRD — see #366's "Out of Scope".
+
+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: ["pgsql_records", "pgsql_files", "rustfs", "raw_file"]
+    pgsql_records:
+      type: postgres
+      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"]
+      # dc_e2e_deg_postgres, not 127.0.0.1: run_degraded.sh's shaped bridge network
+      # resolves it via container-name DNS (podman's aardvark-dns on user-defined
+      # networks) — see this file's header.
+      host: "dc_e2e_deg_postgres"
+      port: 5432
+      user: "dc"
+      password: "password"
+      database: "dc"
+      table: "dc_records"
+      time_key: "date"
+
+    pgsql_files:
+      type: postgres
+      receives: records
+      host: "dc_e2e_deg_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"
+      endpoint: "http://dc_e2e_deg_rustfs:9000"
+      region: "us-east-1"
+      access_key_id: "rustfsadmin"
+      secret_access_key: "rustfsadmin"
+      force_path_style: true
+
+    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"]
+
+    # Same-container loopback (Bridge-to-Shipper) — not shaped, see this file's header.
+    vector_forward_host: "127.0.0.1"
+    vector_forward_port: 24224
+
+lifecycle_manager_dc:
+  ros__parameters:
+    node_names: ["measurement_server"]
+    transitions: [configure, activate]