From 19789cfde3219b46858f24bcde52350b6178f25a Mon Sep 17 00:00:00 2001 From: Cristi Bleotiu Date: Tue, 18 Aug 2026 09:11:58 +0300 Subject: [PATCH 1/4] feat: configure staged CommandControl rollout What changed: - configure regrouped communication roles and heartbeat ingress controls - add targeted heartbeat testbed support and legacy rollback fixtures - document persisted config migration, rollout, and SDK fanout validation Why: - deploy the CommandControl changes compatibly across existing and new nodes - provide executable broker and configuration rollout evidence --- .config_app.json | 12 +- .config_app_cluster.json | 16 +- .config_app_comms.json | 12 +- .../config_app_deeploy_testbed.json | 14 +- docker-compose_comms.yaml | 1 + docs/heartbeat-ingress-rollout.md | 96 +++++ docs/heartbeat-observation-rollout.md | 54 +++ .../communication_topology_legacy.json | 10 + tests/fixtures/config_app_comms_legacy.json | 77 ++++ tests/test_commandcontrol_combined_config.py | 82 +++++ tests/test_comms_testbed_config.py | 61 +++- tests/validate_comms_broker_state.py | 62 +++- tests/validate_sdk_heartbeat_fanout.py | 329 ++++++++++++++++++ 13 files changed, 801 insertions(+), 25 deletions(-) create mode 100644 docs/heartbeat-ingress-rollout.md create mode 100644 docs/heartbeat-observation-rollout.md create mode 100644 tests/fixtures/communication_topology_legacy.json create mode 100644 tests/fixtures/config_app_comms_legacy.json create mode 100644 tests/test_commandcontrol_combined_config.py create mode 100644 tests/validate_sdk_heartbeat_fanout.py diff --git a/.config_app.json b/.config_app.json index eaf6f847..57299d45 100644 --- a/.config_app.json +++ b/.config_app.json @@ -2,14 +2,14 @@ "COMMUNICATION": { "INSTANCES": { "COMMANDCONTROL": { - "RECV_FROM": "CTRL_CHANNEL", + "RECV_FROM": "CONFIG_CHANNEL", "SEND_TO": "CONFIG_CHANNEL" }, "DEFAULT": { "SEND_TO": "PAYLOADS_CHANNEL" }, "HEARTBEATS": { - "RECV_FROM": "CONFIG_CHANNEL", + "RECV_FROM": "CTRL_CHANNEL", "SEND_TO": "CTRL_CHANNEL" }, "NOTIFICATIONS": { @@ -31,9 +31,17 @@ }, "CTRL_CHANNEL": { "TOPIC": "naeural/ctrl", + "TARGETED_TOPIC": "naeural/ctrl/{}", + "SUBSCRIBE_TARGETED": false, "QOS": 1 }, "HOST": "$EE_MQTT_HOST", + "HEARTBEAT_AUTH_MODE": "shadow", + "HEARTBEAT_AUTH_MAX_IN_FLIGHT": 32, + "HEARTBEAT_AUTH_WORKERS": 4, + "HEARTBEAT_INGRESS_QUEUE_SIZE": 10000, + "HEARTBEAT_INGRESS_WORKER_ENABLED": true, + "HEARTBEAT_TARGETED_MIRROR_ENABLED": false, "NOTIF_CHANNEL": { "TOPIC": "naeural/notif" }, diff --git a/.config_app_cluster.json b/.config_app_cluster.json index 0c30f32a..d03d254e 100644 --- a/.config_app_cluster.json +++ b/.config_app_cluster.json @@ -2,14 +2,14 @@ "COMMUNICATION": { "INSTANCES": { "COMMANDCONTROL": { - "RECV_FROM": "CTRL_CHANNEL", + "RECV_FROM": "CONFIG_CHANNEL", "SEND_TO": "CONFIG_CHANNEL" }, "DEFAULT": { "SEND_TO": "PAYLOADS_CHANNEL" }, "HEARTBEATS": { - "RECV_FROM": "CONFIG_CHANNEL", + "RECV_FROM": "CTRL_CHANNEL", "SEND_TO": "CTRL_CHANNEL" }, "NOTIFICATIONS": { @@ -29,9 +29,17 @@ "TOPIC": "naeural_test/{}/config" }, "CTRL_CHANNEL": { - "TOPIC": "naeural_test/ctrl" + "TOPIC": "naeural_test/ctrl", + "TARGETED_TOPIC": "naeural_test/ctrl/{}", + "SUBSCRIBE_TARGETED": false }, "HOST": "$EE_MQTT_HOST", + "HEARTBEAT_AUTH_MODE": "shadow", + "HEARTBEAT_AUTH_MAX_IN_FLIGHT": 32, + "HEARTBEAT_AUTH_WORKERS": 4, + "HEARTBEAT_INGRESS_QUEUE_SIZE": 10000, + "HEARTBEAT_INGRESS_WORKER_ENABLED": true, + "HEARTBEAT_TARGETED_MIRROR_ENABLED": false, "NOTIF_CHANNEL": { "TOPIC": "naeural_test/notif" }, @@ -62,4 +70,4 @@ "PAYLOAD_TYPE": "INSTANCE", "SECONDS_HEARTBEAT": 10, "SERVING_IN_PROCESS": true -} \ No newline at end of file +} diff --git a/.config_app_comms.json b/.config_app_comms.json index 565b7b92..c282e60a 100644 --- a/.config_app_comms.json +++ b/.config_app_comms.json @@ -2,14 +2,14 @@ "COMMUNICATION": { "INSTANCES": { "COMMANDCONTROL": { - "RECV_FROM": "CTRL_CHANNEL", + "RECV_FROM": "CONFIG_CHANNEL", "SEND_TO": "CONFIG_CHANNEL" }, "DEFAULT": { "SEND_TO": "PAYLOADS_CHANNEL" }, "HEARTBEATS": { - "RECV_FROM": "CONFIG_CHANNEL", + "RECV_FROM": "CTRL_CHANNEL", "SEND_TO": "CTRL_CHANNEL" }, "NOTIFICATIONS": { @@ -31,9 +31,17 @@ }, "CTRL_CHANNEL": { "TOPIC": "naeural_comms_local/ctrl", + "TARGETED_TOPIC": "naeural_comms_local/ctrl/{}", + "SUBSCRIBE_TARGETED": false, "QOS": 1 }, "HOST": "emqx", + "HEARTBEAT_AUTH_MODE": "shadow", + "HEARTBEAT_AUTH_MAX_IN_FLIGHT": 32, + "HEARTBEAT_AUTH_WORKERS": 4, + "HEARTBEAT_INGRESS_QUEUE_SIZE": 10000, + "HEARTBEAT_INGRESS_WORKER_ENABLED": true, + "HEARTBEAT_TARGETED_MIRROR_ENABLED": true, "NOTIF_CHANNEL": { "TOPIC": "naeural_comms_local/notif", "QOS": 0 diff --git a/docker-compose/deeploy-testbed/config_app_deeploy_testbed.json b/docker-compose/deeploy-testbed/config_app_deeploy_testbed.json index 02497f38..efcc28df 100644 --- a/docker-compose/deeploy-testbed/config_app_deeploy_testbed.json +++ b/docker-compose/deeploy-testbed/config_app_deeploy_testbed.json @@ -2,14 +2,14 @@ "COMMUNICATION": { "INSTANCES": { "COMMANDCONTROL": { - "RECV_FROM": "CTRL_CHANNEL", + "RECV_FROM": "CONFIG_CHANNEL", "SEND_TO": "CONFIG_CHANNEL" }, "DEFAULT": { "SEND_TO": "PAYLOADS_CHANNEL" }, "HEARTBEATS": { - "RECV_FROM": "CONFIG_CHANNEL", + "RECV_FROM": "CTRL_CHANNEL", "SEND_TO": "CTRL_CHANNEL" }, "NOTIFICATIONS": { @@ -29,9 +29,17 @@ "TOPIC": "deeploy_testbed/{}/config" }, "CTRL_CHANNEL": { - "TOPIC": "deeploy_testbed/ctrl" + "TOPIC": "deeploy_testbed/ctrl", + "TARGETED_TOPIC": "deeploy_testbed/ctrl/{}", + "SUBSCRIBE_TARGETED": false }, "HOST": "$EE_MQTT_HOST", + "HEARTBEAT_AUTH_MODE": "shadow", + "HEARTBEAT_AUTH_MAX_IN_FLIGHT": 32, + "HEARTBEAT_AUTH_WORKERS": 4, + "HEARTBEAT_INGRESS_QUEUE_SIZE": 10000, + "HEARTBEAT_INGRESS_WORKER_ENABLED": true, + "HEARTBEAT_TARGETED_MIRROR_ENABLED": false, "NOTIF_CHANNEL": { "TOPIC": "deeploy_testbed/notif" }, diff --git a/docker-compose_comms.yaml b/docker-compose_comms.yaml index 6c193b93..de40de75 100644 --- a/docker-compose_comms.yaml +++ b/docker-compose_comms.yaml @@ -27,6 +27,7 @@ x-edge-comms-node: &edge-comms-node EE_MQTT_SUBTOPIC: address EE_MQTT_HEARTBEAT_QOS: "1" EE_MQTT_COMMAND_QOS: "2" + EE_HEARTBEAT_TARGETED_MIRROR_ENABLED: "1" EE_NETMON_ORACLE_ONLY_HEARTBEAT_MODE: "1" EE_ENABLE_NETMON_API_PROBE: "1" # The comms testbed does not run child workloads, so DinD TLS only adds a diff --git a/docs/heartbeat-ingress-rollout.md b/docs/heartbeat-ingress-rollout.md new file mode 100644 index 00000000..afd48782 --- /dev/null +++ b/docs/heartbeat-ingress-rollout.md @@ -0,0 +1,96 @@ +# Heartbeat Ingress Rollout + +The runtime now admits raw CTRL heartbeats into a bounded in-memory FIFO, +authenticates signed envelopes on a small worker set, and commits formatter, +deduplication, NetMon, and Epoch state on one ordered owner thread. + +## Runtime Configuration + +The recommended initial settings are: + +```json +{ + "HEARTBEAT_INGRESS_WORKER_ENABLED": true, + "HEARTBEAT_INGRESS_QUEUE_SIZE": 10000, + "HEARTBEAT_AUTH_WORKERS": 4, + "HEARTBEAT_AUTH_MAX_IN_FLIGHT": 32, + "HEARTBEAT_AUTH_MODE": "shadow" +} +``` + +`HEARTBEAT_AUTH_WORKERS` performs only raw JSON parsing and signature +verification concurrently. `HEARTBEAT_AUTH_MAX_IN_FLIGHT` bounds the work that +has left the main FIFO but is waiting for its ordered turn. Formatter decoding, +deduplication, NetMon, and Epoch mutation remain serial and preserve FIFO order. +Signing-canonicalization counters remain in memory during heartbeat processing; +their best-effort JSON snapshot is flushed after the heartbeat worker drains on +clean shutdown, so metric persistence does not compete with authentication. If +the worker exceeds its shutdown timeout, runtime status and logs report an +incomplete drain and the snapshot is not flushed; shutdown persistence is +best-effort rather than absolute. + +The regrouped communication roles are: + +```json +{ + "COMMANDCONTROL": { + "RECV_FROM": "CONFIG_CHANNEL", + "SEND_TO": "CONFIG_CHANNEL" + }, + "HEARTBEATS": { + "RECV_FROM": "CTRL_CHANNEL", + "SEND_TO": "CTRL_CHANNEL" + } +} +``` + +The old crossed roles remain supported for rollback. Receive behavior is +derived from the configured channel, not the communicator name. + +## Existing Nodes And `config_app.txt` + +Existing nodes must receive the new settings and regrouped roles in their +effective persisted configuration. The tracked `.config_app*.json` files seed +new or reset installations. In the communication testbed that seed is +`.config_app_comms.json`, while a running node normally loads: + +```text +/edge_node/_local_cache/_data/box_configuration/config_app.txt +``` + +Update that file through the normal configuration distribution mechanism; do +not commit a generated `config_app.txt`. A rollback test must update both the +startup seed, such as `.config_app_comms.json`, and the persisted +`config_app.txt`; otherwise startup can immediately restore the seeded values. +Verify the effective persisted file after restart or configuration reload. + +## Safe Sequence + +1. Deploy the compatible code and keep the legacy topology with + `HEARTBEAT_AUTH_MODE=shadow`. +2. Confirm conservation, rejected-full, oldest-age, authentication, and commit + counters remain healthy for at least one full Epoch boundary. +3. Change the persisted communication roles to the regrouped topology and + verify broker subscriptions and command delivery. +4. Move authentication to `enforce` only after shadow evidence accounts for + legacy senders and present identity mismatches. + +Rollback is configuration-only: restore the legacy role mapping or set +`HEARTBEAT_INGRESS_WORKER_ENABLED=false`. With the worker disabled, whichever +communicator owns CTRL continues consuming one heartbeat at a time on its own +loop through the same authentication and identity checks. This fallback works +with both legacy and regrouped roles, but it restores the old throughput +coupling, so use it as an emergency rollback rather than a steady-state tuning +choice. Do not change both topology and auth enforcement in the same rollout +step. + +## Capacity And Durability + +When the queue is full, the new message is rejected and counted; already +accepted FIFO entries are not evicted. Queue depth is the number waiting in the +FIFO. Oldest age reads the timestamp on the head entry in constant time. + +This release has no write-ahead log. A broker disconnect does not erase the +process-local queue while the process remains alive, but a process crash can +lose admitted messages. The counters make that loss boundary visible; they do +not provide durable replay or exactly-once delivery. diff --git a/docs/heartbeat-observation-rollout.md b/docs/heartbeat-observation-rollout.md new file mode 100644 index 00000000..728048b4 --- /dev/null +++ b/docs/heartbeat-observation-rollout.md @@ -0,0 +1,54 @@ +# Heartbeat Observation Rollout + +The SDK supports three immutable heartbeat observation modes: + +- `full_network` keeps the existing global `CTRL` subscription and is the default. +- `selected_nodes` subscribes only to exact `CTRL/` topics. +- `summary_discovery` does not subscribe to `CTRL`; it discovers network state from trusted, signed `NET_MON_01` payloads. + +## Runtime Configuration + +The edge runtime needs this communication shape before any selected-node SDK is enabled: + +```json +{ + "CTRL_CHANNEL": { + "TOPIC": "naeural/ctrl", + "TARGETED_TOPIC": "naeural/ctrl/{}", + "SUBSCRIBE_TARGETED": false + }, + "HEARTBEAT_TARGETED_MIRROR_ENABLED": false +} +``` + +Use the deployment's actual topic root. `SUBSCRIBE_TARGETED` remains `false` for edge communicators because the targeted route is for SDK consumers; subscribing the edge CommandControl receiver to its own route would duplicate heartbeat processing. + +When `HEARTBEAT_TARGETED_MIRROR_ENABLED` is enabled, a node publishes the same serialized, signed heartbeat once to the global topic and once to its own addressed topic. The flag can also be supplied as `EE_HEARTBEAT_TARGETED_MIRROR_ENABLED=true`. The default remains global-only. + +## Existing Nodes And `config_app.txt` + +Yes, existing nodes need their persisted application configuration updated. The tracked `.config_app*.json` files seed new or reset installations. In the communication testbed that seed is `.config_app_comms.json`, but a running node normally loads: + +```text +/edge_node/_local_cache/_data/box_configuration/config_app.txt +``` + +Update that persisted configuration through the normal configuration distribution or deployment mechanism so `CTRL_CHANNEL.TARGETED_TOPIC` and `CTRL_CHANNEL.SUBSCRIBE_TARGETED` are present. Do not add a generated `config_app.txt` to source control. Rollback tests must update both `.config_app_comms.json` and the persisted file so startup seeding cannot undo the rollback. After update and restart or configuration reload, verify the effective persisted file before enabling the mirror. + +If the mirror flag is enabled while `TARGETED_TOPIC` is absent, the runtime warns once and continues publishing globally. It does not invent a topic. A `selected_nodes` SDK will therefore remain degraded instead of silently subscribing to the full network. + +## Safe Sequence + +1. Deploy the runtime and SDK code while all clients remain in the default `full_network` mode. +2. Add the targeted topic with the mirror disabled to templates and existing nodes' persisted configuration. +3. Verify the effective config and the unchanged global heartbeat path. +4. Enable the heartbeat mirror on a small node cohort. +5. Start `selected_nodes` SDK sessions for those nodes and require `state == "ready"` from `get_heartbeat_observation_status()`. +6. Start `summary_discovery` only with explicitly trusted NetMon publisher addresses and require a fresh accepted summary. +7. Expand only while broker delivery counters, SDK queue counters, and observation freshness remain healthy. + +Rollback is configuration-only: recreate reduced-mode SDK sessions as `full_network`, then disable the mirror. Leaving `TARGETED_TOPIC` configured while the mirror is disabled is harmless. + +## Durability Boundary + +This change does not add a write-ahead log. Messages already admitted to an in-memory SDK or node queue survive a broker disconnect while the process remains alive, but a process crash can lose them. Queue conservation and freshness metrics make that risk visible; they do not provide durable replay. diff --git a/tests/fixtures/communication_topology_legacy.json b/tests/fixtures/communication_topology_legacy.json new file mode 100644 index 00000000..2e745b69 --- /dev/null +++ b/tests/fixtures/communication_topology_legacy.json @@ -0,0 +1,10 @@ +{ + "COMMANDCONTROL": { + "RECV_FROM": "CTRL_CHANNEL", + "SEND_TO": "CONFIG_CHANNEL" + }, + "HEARTBEATS": { + "RECV_FROM": "CONFIG_CHANNEL", + "SEND_TO": "CTRL_CHANNEL" + } +} diff --git a/tests/fixtures/config_app_comms_legacy.json b/tests/fixtures/config_app_comms_legacy.json new file mode 100644 index 00000000..5bf9f069 --- /dev/null +++ b/tests/fixtures/config_app_comms_legacy.json @@ -0,0 +1,77 @@ +{ + "COMMUNICATION": { + "INSTANCES": { + "COMMANDCONTROL": { + "RECV_FROM": "CTRL_CHANNEL", + "SEND_TO": "CONFIG_CHANNEL" + }, + "DEFAULT": { + "SEND_TO": "PAYLOADS_CHANNEL" + }, + "HEARTBEATS": { + "RECV_FROM": "CONFIG_CHANNEL", + "SEND_TO": "CTRL_CHANNEL" + }, + "NOTIFICATIONS": { + "SEND_TO": "NOTIF_CHANNEL" + } + }, + "LOCAL_PARAMS": { + "HOST": "emqx", + "PASS": "", + "PORT": 1883, + "USER": "", + "QOS": 0, + "SECURED": 0 + }, + "PARAMS": { + "CONFIG_CHANNEL": { + "TOPIC": "naeural_comms_local/{}/config", + "QOS": 2 + }, + "CTRL_CHANNEL": { + "TOPIC": "naeural_comms_local/ctrl", + "TARGETED_TOPIC": "naeural_comms_local/ctrl/{}", + "SUBSCRIBE_TARGETED": false, + "QOS": 1 + }, + "HOST": "emqx", + "HEARTBEAT_AUTH_MODE": "shadow", + "HEARTBEAT_AUTH_MAX_IN_FLIGHT": 32, + "HEARTBEAT_AUTH_WORKERS": 4, + "HEARTBEAT_INGRESS_QUEUE_SIZE": 10000, + "HEARTBEAT_INGRESS_WORKER_ENABLED": true, + "HEARTBEAT_TARGETED_MIRROR_ENABLED": false, + "NOTIF_CHANNEL": { + "TOPIC": "naeural_comms_local/notif", + "QOS": 0 + }, + "PASS": "", + "PAYLOADS_CHANNEL": { + "TOPIC": "naeural_comms_local/payloads", + "TARGETED_TOPIC": "naeural_comms_local/{}/payloads", + "QOS": 0 + }, + "PORT": 1883, + "QOS": 0, + "USER": "", + "SUBTOPIC": "address", + "SECURED": 0, + "CERT_PATH": "" + }, + "TYPE": "mqtt" + }, + "FILE_UPLOAD": { + "CONFIG_UPLOADER": { + "ACCESS_KEY": "", + "BUCKET_NAME": "", + "ENDPOINT": "", + "SECRET_KEY": "", + "SECURE": false + }, + "TYPE": "minio" + }, + "PAYLOAD_TYPE": "INSTANCE", + "SECONDS_HEARTBEAT": 5, + "SERVING_IN_PROCESS": true +} diff --git a/tests/test_commandcontrol_combined_config.py b/tests/test_commandcontrol_combined_config.py new file mode 100644 index 00000000..148f03e1 --- /dev/null +++ b/tests/test_commandcontrol_combined_config.py @@ -0,0 +1,82 @@ +import json +import unittest +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +class TestCombinedCommandControlConfig(unittest.TestCase): + + def _load(self, relative_path): + return json.loads((REPO_ROOT / relative_path).read_text()) + + def _assert_regrouped(self, config): + instances = config["COMMUNICATION"]["INSTANCES"] + self.assertEqual(instances["COMMANDCONTROL"]["RECV_FROM"], "CONFIG_CHANNEL") + self.assertEqual(instances["COMMANDCONTROL"]["SEND_TO"], "CONFIG_CHANNEL") + self.assertEqual(instances["HEARTBEATS"]["RECV_FROM"], "CTRL_CHANNEL") + self.assertEqual(instances["HEARTBEATS"]["SEND_TO"], "CTRL_CHANNEL") + + def _assert_combined_params(self, config, mirror_enabled): + params = config["COMMUNICATION"]["PARAMS"] + self.assertTrue(params["HEARTBEAT_INGRESS_WORKER_ENABLED"]) + self.assertEqual(params["HEARTBEAT_INGRESS_QUEUE_SIZE"], 10000) + self.assertEqual(params["HEARTBEAT_AUTH_WORKERS"], 4) + self.assertEqual(params["HEARTBEAT_AUTH_MAX_IN_FLIGHT"], 32) + self.assertEqual(params["HEARTBEAT_AUTH_MODE"], "shadow") + self.assertIn("{}", params["CTRL_CHANNEL"]["TARGETED_TOPIC"]) + self.assertFalse(params["CTRL_CHANNEL"]["SUBSCRIBE_TARGETED"]) + self.assertEqual( + params["HEARTBEAT_TARGETED_MIRROR_ENABLED"], + mirror_enabled, + ) + + def test_comms_testbed_is_regrouped_with_opt_in_targeted_mirror(self): + config = self._load(".config_app_comms.json") + self._assert_regrouped(config) + self._assert_combined_params(config, mirror_enabled=True) + + def test_tracked_runtime_defaults_are_regrouped_but_mirror_off(self): + for relative_path in ( + ".config_app.json", + ".config_app_cluster.json", + "docker-compose/deeploy-testbed/config_app_deeploy_testbed.json", + ): + with self.subTest(relative_path=relative_path): + config = self._load(relative_path) + self._assert_regrouped(config) + self._assert_combined_params(config, mirror_enabled=False) + + def test_legacy_fixture_preserves_crossed_roles_for_config_rollback(self): + fixture = self._load("tests/fixtures/config_app_comms_legacy.json") + instances = fixture["COMMUNICATION"]["INSTANCES"] + self.assertEqual(instances["COMMANDCONTROL"]["RECV_FROM"], "CTRL_CHANNEL") + self.assertEqual(instances["HEARTBEATS"]["RECV_FROM"], "CONFIG_CHANNEL") + self._assert_combined_params(fixture, mirror_enabled=False) + + def test_rollout_docs_name_seed_and_persisted_config_boundaries(self): + ingress = (REPO_ROOT / "docs" / "heartbeat-ingress-rollout.md").read_text() + observation = ( + REPO_ROOT / "docs" / "heartbeat-observation-rollout.md" + ).read_text() + + for text in (ingress, observation): + self.assertIn("config_app.txt", text) + self.assertIn(".config_app_comms.json", text) + self.assertIn("rollback", text.lower()) + + def test_live_testbed_enables_mirror_and_has_broker_fanout_probe(self): + compose = (REPO_ROOT / "docker-compose_comms.yaml").read_text() + probe = (REPO_ROOT / "tests" / "validate_sdk_heartbeat_fanout.py").read_text() + + self.assertIn('EE_HEARTBEAT_TARGETED_MIRROR_ENABLED: "1"', compose) + self.assertIn("/api/v5/subscriptions", probe) + self.assertIn("send_msg", probe) + self.assertIn("send_oct", probe) + self.assertIn("post_delivery_filter", probe) + self.assertIn("heartbeat_observation_mode", probe) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_comms_testbed_config.py b/tests/test_comms_testbed_config.py index 67635136..bd38a2f8 100644 --- a/tests/test_comms_testbed_config.py +++ b/tests/test_comms_testbed_config.py @@ -42,16 +42,21 @@ def test_compose_has_supervisors_and_normal_nodes(self): self.assertIn("EE_NETMON_ACCEPT_LOCAL_SUPERVISOR_SUMMARY: \"1\"", compose_text) self.assertNotIn("EE_NETMON_USE_SUMMARY_STATUS", compose_text) - def test_app_config_has_isolated_topics_and_channel_qos(self): + def test_app_config_has_regrouped_traffic_roles_and_channel_qos(self): config = json.loads((REPO_ROOT / ".config_app_comms.json").read_text()) instances = config["COMMUNICATION"]["INSTANCES"] params = config["COMMUNICATION"]["PARAMS"] - self.assertEqual(instances["COMMANDCONTROL"]["RECV_FROM"], "CTRL_CHANNEL") + self.assertEqual(instances["COMMANDCONTROL"]["RECV_FROM"], "CONFIG_CHANNEL") self.assertEqual(instances["COMMANDCONTROL"]["SEND_TO"], "CONFIG_CHANNEL") - self.assertEqual(instances["HEARTBEATS"]["RECV_FROM"], "CONFIG_CHANNEL") + self.assertEqual(instances["HEARTBEATS"]["RECV_FROM"], "CTRL_CHANNEL") self.assertEqual(instances["HEARTBEATS"]["SEND_TO"], "CTRL_CHANNEL") self.assertEqual(params["HOST"], "emqx") + self.assertEqual(params["HEARTBEAT_AUTH_MODE"], "shadow") + self.assertEqual(params["HEARTBEAT_AUTH_MAX_IN_FLIGHT"], 32) + self.assertEqual(params["HEARTBEAT_AUTH_WORKERS"], 4) + self.assertEqual(params["HEARTBEAT_INGRESS_QUEUE_SIZE"], 10000) + self.assertTrue(params["HEARTBEAT_INGRESS_WORKER_ENABLED"]) self.assertEqual(params["PORT"], 1883) self.assertEqual(params["SECURED"], 0) self.assertEqual(params["CTRL_CHANNEL"]["TOPIC"], "naeural_comms_local/ctrl") @@ -63,13 +68,63 @@ def test_app_config_has_isolated_topics_and_channel_qos(self): def test_default_app_config_enables_segregated_heartbeat_and_command_qos(self): config = json.loads((REPO_ROOT / ".config_app.json").read_text()) + instances = config["COMMUNICATION"]["INSTANCES"] params = config["COMMUNICATION"]["PARAMS"] + self.assertEqual(instances["COMMANDCONTROL"]["RECV_FROM"], "CONFIG_CHANNEL") + self.assertEqual(instances["COMMANDCONTROL"]["SEND_TO"], "CONFIG_CHANNEL") + self.assertEqual(instances["HEARTBEATS"]["RECV_FROM"], "CTRL_CHANNEL") + self.assertEqual(instances["HEARTBEATS"]["SEND_TO"], "CTRL_CHANNEL") self.assertEqual(params["CTRL_CHANNEL"]["TOPIC"], "naeural/ctrl") self.assertEqual(params["CTRL_CHANNEL"]["QOS"], 1) self.assertEqual(params["CONFIG_CHANNEL"]["TOPIC"], "naeural/{}/config") self.assertEqual(params["CONFIG_CHANNEL"]["QOS"], 2) self.assertEqual(params["QOS"], 2) + self.assertEqual(params["HEARTBEAT_AUTH_MODE"], "shadow") + self.assertEqual(params["HEARTBEAT_AUTH_MAX_IN_FLIGHT"], 32) + self.assertEqual(params["HEARTBEAT_AUTH_WORKERS"], 4) + self.assertEqual(params["HEARTBEAT_INGRESS_QUEUE_SIZE"], 10000) + self.assertTrue(params["HEARTBEAT_INGRESS_WORKER_ENABLED"]) + + def test_legacy_topology_fixture_remains_available_for_rollback(self): + fixture = json.loads(( + REPO_ROOT / "tests" / "fixtures" / "communication_topology_legacy.json" + ).read_text()) + + self.assertEqual(fixture["COMMANDCONTROL"]["RECV_FROM"], "CTRL_CHANNEL") + self.assertEqual(fixture["COMMANDCONTROL"]["SEND_TO"], "CONFIG_CHANNEL") + self.assertEqual(fixture["HEARTBEATS"]["RECV_FROM"], "CONFIG_CHANNEL") + self.assertEqual(fixture["HEARTBEATS"]["SEND_TO"], "CTRL_CHANNEL") + + full_fixture = json.loads(( + REPO_ROOT / "tests" / "fixtures" / "config_app_comms_legacy.json" + ).read_text()) + full_instances = full_fixture["COMMUNICATION"]["INSTANCES"] + self.assertEqual(full_instances["COMMANDCONTROL"], fixture["COMMANDCONTROL"]) + self.assertEqual(full_instances["HEARTBEATS"], fixture["HEARTBEATS"]) + self.assertEqual( + full_fixture["COMMUNICATION"]["PARAMS"]["CTRL_CHANNEL"]["QOS"], + 1, + ) + self.assertEqual( + full_fixture["COMMUNICATION"]["PARAMS"]["CONFIG_CHANNEL"]["QOS"], + 2, + ) + self.assertEqual( + full_fixture["COMMUNICATION"]["PARAMS"]["HEARTBEAT_AUTH_WORKERS"], + 4, + ) + + def test_heartbeat_ingress_rollout_documents_persisted_config_and_rollback(self): + rollout = ( + REPO_ROOT / "docs" / "heartbeat-ingress-rollout.md" + ).read_text() + + self.assertIn("config_app.txt", rollout) + self.assertIn("HEARTBEAT_AUTH_WORKERS", rollout) + self.assertIn("HEARTBEAT_AUTH_MAX_IN_FLIGHT", rollout) + self.assertIn("HEARTBEAT_INGRESS_WORKER_ENABLED=false", rollout) + self.assertIn("no write-ahead log", rollout.lower()) def test_startup_config_keeps_required_admin_pipeline_only(self): config = json.loads((REPO_ROOT / ".config_startup_comms.json").read_text()) diff --git a/tests/validate_comms_broker_state.py b/tests/validate_comms_broker_state.py index 64b7575d..57d314b5 100644 --- a/tests/validate_comms_broker_state.py +++ b/tests/validate_comms_broker_state.py @@ -22,6 +22,15 @@ API_TOKEN = os.environ.get("ECOMMS_EMQX_TOKEN") ROOT_TOPIC = os.environ.get("ECOMMS_ROOT_TOPIC", "naeural_comms_local") CTRL_TOPIC = f"{ROOT_TOPIC}/ctrl" +EXPECTED_TOPOLOGY = os.environ.get( + "ECOMMS_EXPECTED_TOPOLOGY", "regrouped", +).strip().lower() +if EXPECTED_TOPOLOGY not in {"legacy", "regrouped"}: + raise ValueError( + "ECOMMS_EXPECTED_TOPOLOGY must be 'legacy' or 'regrouped', got {!r}".format( + EXPECTED_TOPOLOGY, + ) + ) EXPECTED_NODES = { "comm_oracle_01": "supervisor", "comm_oracle_02": "supervisor", @@ -166,13 +175,27 @@ def _validate_state(subscriptions, clients): row for row in alias_subs if alias in _client_id(row) and row.get("topic") == CTRL_TOPIC ] - command_ctrl_subs = [ + heartbeat_receiver_marker = ( + "_HEARTBE" if EXPECTED_TOPOLOGY == "regrouped" else "_COMMAND" + ) + command_receiver_marker = ( + "_COMMAND" if EXPECTED_TOPOLOGY == "regrouped" else "_HEARTBE" + ) + heartbeat_ctrl_subs = [ row for row in alias_subs - if alias in _client_id(row) and "_COMMAND" in _client_id(row) and row.get("topic") == CTRL_TOPIC + if ( + alias in _client_id(row) + and heartbeat_receiver_marker in _client_id(row) + and row.get("topic") == CTRL_TOPIC + ) ] - heartbeat_config_subs = [ + command_config_subs = [ row for row in alias_subs - if alias in _client_id(row) and "_HEARTBE" in _client_id(row) and str(row.get("topic", "")).endswith("/config") + if ( + alias in _client_id(row) + and command_receiver_marker in _client_id(row) + and str(row.get("topic", "")).endswith("/config") + ) ] iot_payload_broadcast_subs = [ row for row in alias_subs @@ -195,17 +218,33 @@ def _validate_state(subscriptions, clients): ) if role == "supervisor": - _assert(len(command_ctrl_subs) == 1, f"{alias}: expected exactly one COMMAND ctrl subscription", failures) + _assert( + len(heartbeat_ctrl_subs) == 1, + f"{alias}: expected exactly one {heartbeat_receiver_marker} ctrl subscription", + failures, + ) _assert(len(ctrl_subs) == 1, f"{alias}: expected exactly one total ctrl subscription, found {len(ctrl_subs)}", failures) - if command_ctrl_subs: - _assert(_qos(command_ctrl_subs[0]) == 1, f"{alias}: COMMAND ctrl QoS is not 1", failures) + if heartbeat_ctrl_subs: + _assert( + _qos(heartbeat_ctrl_subs[0]) == 1, + f"{alias}: {heartbeat_receiver_marker} ctrl QoS is not 1", + failures, + ) else: - _assert(len(command_ctrl_subs) == 0, f"{alias}: normal node must not subscribe COMMAND to ctrl", failures) + _assert(len(heartbeat_ctrl_subs) == 0, f"{alias}: normal node must not subscribe HEARTBEAT to ctrl", failures) _assert(len(ctrl_subs) == 0, f"{alias}: normal node must not have any ctrl subscription: {[_client_id(row) for row in ctrl_subs]}", failures) - _assert(len(heartbeat_config_subs) == 1, f"{alias}: expected exactly one HEARTBE config subscription", failures) - if heartbeat_config_subs: - _assert(_qos(heartbeat_config_subs[0]) == 2, f"{alias}: HEARTBE config QoS is not 2", failures) + _assert( + len(command_config_subs) == 1, + f"{alias}: expected exactly one {command_receiver_marker} config subscription", + failures, + ) + if command_config_subs: + _assert( + _qos(command_config_subs[0]) == 2, + f"{alias}: {command_receiver_marker} config QoS is not 2", + failures, + ) _assert(len(iot_payload_broadcast_subs) == 1, f"{alias}: expected exactly one IoT broadcast payload subscription", failures) if iot_payload_broadcast_subs: _assert(_qos(iot_payload_broadcast_subs[0]) == 0, f"{alias}: IoT broadcast payload QoS is not 0", failures) @@ -256,6 +295,7 @@ def main(): "subscriptions_checked": len(subscriptions), "managed_clients_checked": len(managed_clients), "ctrl_topic": CTRL_TOPIC, + "topology": EXPECTED_TOPOLOGY, "status": "ok", }, indent=2)) return 0 diff --git a/tests/validate_sdk_heartbeat_fanout.py b/tests/validate_sdk_heartbeat_fanout.py new file mode 100644 index 00000000..8639810c --- /dev/null +++ b/tests/validate_sdk_heartbeat_fanout.py @@ -0,0 +1,329 @@ +#!/usr/bin/env python3 +"""Prove SDK heartbeat modes by broker delivery, not post-delivery filtering. + +Run this only against ``docker-compose_comms.yaml``. The probe discovers one +node and one NetMon publisher, starts real Ratio1 SDK sessions for every mode, +then compares EMQX per-client delivered messages and bytes over one shared +traffic window. +""" + +import importlib.util +import json +import os +import pathlib +import tempfile +import threading +import time + +import paho.mqtt.client as mqtt + +from ratio1 import Session + + +ROOT_TOPIC = os.environ.get("ECOMMS_ROOT_TOPIC", "naeural_comms_local") +HOST = os.environ.get("ECOMMS_MQTT_HOST", "127.0.0.1") +PORT = int(os.environ.get("ECOMMS_MQTT_PORT", "18883")) +# The isolated broker accepts anonymous MQTT, but the public SDK deliberately +# requires non-empty credential fields before connecting. EMQX accepts these +# local-only placeholders while deployments can override both values. +USER = os.environ.get("ECOMMS_MQTT_USER", "ecomms") +PASSWORD = os.environ.get("ECOMMS_MQTT_PASS", "ecomms") +DISCOVERY_SECONDS = float(os.environ.get("ECOMMS_SDK_DISCOVERY_SECONDS", "90")) +MEASURE_SECONDS = float(os.environ.get("ECOMMS_SDK_MEASURE_SECONDS", "30")) + + +def _broker_module(): + path = pathlib.Path(__file__).with_name("validate_comms_broker_state.py") + spec = importlib.util.spec_from_file_location("_ecomms_broker", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _mqtt_client(client_id): + if hasattr(mqtt, "CallbackAPIVersion"): + client = mqtt.Client( + callback_api_version=mqtt.CallbackAPIVersion.VERSION2, + client_id=client_id, + clean_session=True, + ) + else: + client = mqtt.Client(client_id=client_id, clean_session=True) + client.username_pw_set(USER, PASSWORD) + return client + + +def _discover_publishers(): + state = { + "node": None, + "summary_publisher": None, + } + ready = threading.Event() + client = _mqtt_client("ecomms_sdk_discovery") + + def on_connect(connected_client, userdata, flags, reason_code, *args): + reason_value = getattr(reason_code, "value", reason_code) + if reason_value == 0: + connected_client.subscribe([ + (f"{ROOT_TOPIC}/ctrl", 1), + (f"{ROOT_TOPIC}/payloads", 0), + ]) + + def on_message(_client, _userdata, message): + try: + payload = json.loads(message.payload.decode("utf-8")) + except Exception: + return + sender = payload.get("EE_SENDER") + if message.topic == f"{ROOT_TOPIC}/ctrl" and sender: + state["node"] = state["node"] or sender + path = payload.get("EE_PAYLOAD_PATH", [None, None, None, None]) + if ( + message.topic == f"{ROOT_TOPIC}/payloads" + and len(path) >= 3 + and str(path[1]).lower() == "admin_pipeline" + and str(path[2]).upper() == "NET_MON_01" + and sender + ): + state["summary_publisher"] = state["summary_publisher"] or sender + if all(state.values()): + ready.set() + + client.on_connect = on_connect + client.on_message = on_message + client.connect(HOST, PORT) + client.loop_start() + try: + if not ready.wait(DISCOVERY_SECONDS): + raise RuntimeError( + "Timed out discovering heartbeat and NetMon publishers: {}".format( + state + ) + ) + return state + finally: + client.disconnect() + client.loop_stop() + + +def _session(name, cache_root, **kwargs): + return Session( + host=HOST, + port=PORT, + user=USER, + pwd=PASSWORD, + secured=False, + name=name, + root_topic=ROOT_TOPIC, + auto_configuration=False, + run_dauth=False, + eth_enabled=False, + use_home_folder=False, + local_cache_base_folder=cache_root, + local_cache_app_folder=name, + silent=True, + verbosity=0, + **kwargs, + ) + + +def _find_client(clients, session_name, marker): + matches = [ + row for row in clients + if session_name in str(row.get("clientid", "")) + and marker in str(row.get("clientid", "")) + ] + if len(matches) != 1: + raise AssertionError( + "Expected one client for {} / {}, found {}".format( + session_name, + marker, + [row.get("clientid") for row in matches], + ) + ) + return matches[0] + + +def _delivery(row): + return { + "messages": int(row["send_msg"]), + "bytes": int(row["send_oct"]), + } + + +def _delta(before, after): + return { + key: after[key] - before[key] + for key in before + } + + +def _topics_by_client(subscriptions): + result = {} + for row in subscriptions: + result.setdefault(row.get("clientid", ""), []).append(row.get("topic")) + return result + + +def _session_topics(subscriptions, session_name): + return [ + row.get("topic") + for row in subscriptions + if session_name in str(row.get("clientid", "")) + ] + + +def _start_subscription_sampler(broker, interval=0.25): + samples = [] + errors = [] + stop = threading.Event() + + def sample(): + while not stop.is_set(): + try: + samples.append(broker._fetch_collection("/api/v5/subscriptions")) + except Exception as exc: + errors.append(str(exc)) + stop.wait(interval) + + thread = threading.Thread(target=sample, name="ecomms-subscription-sampler", daemon=True) + thread.start() + return stop, thread, samples, errors + + +def main(): + broker = _broker_module() + discovered = _discover_publishers() + selected_node = discovered["node"] + summary_publisher = discovered["summary_publisher"] + sessions = {} + sampler_stop, sampler_thread, subscription_samples, sampler_errors = ( + _start_subscription_sampler(broker) + ) + + with tempfile.TemporaryDirectory(prefix="ecomms-sdk-fanout-") as cache_root: + try: + sessions["full_network"] = _session("ecomms_full_network", cache_root) + unavailable_node = sessions["full_network"].bc_engine.address + sessions.update({ + "selected_nodes": _session( + "ecomms_selected_nodes", + cache_root, + heartbeat_observation_mode="selected_nodes", + heartbeat_observation_nodes=[selected_node], + ), + "summary_discovery": _session( + "ecomms_summary_discovery", + cache_root, + heartbeat_observation_mode="summary_discovery", + heartbeat_summary_publishers=[summary_publisher], + ), + "post_delivery_filter": _session( + "ecomms_post_delivery_filter", + cache_root, + filter_workers=[selected_node], + ), + "unavailable_selected": _session( + "ecomms_unavailable_selected", + cache_root, + heartbeat_observation_mode="selected_nodes", + heartbeat_observation_nodes=[unavailable_node], + heartbeat_observation_timeout_seconds=5, + ), + }) + + clients = broker._fetch_collection("/api/v5/clients") + before_rows = { + "full_network": _find_client(clients, "ecomms_full_network", "HEARTBE"), + "selected_nodes": _find_client(clients, "ecomms_selected_nodes", "HEARTBE"), + "summary_discovery": _find_client(clients, "ecomms_summary_discovery", "HEARTBE"), + "post_delivery_filter": _find_client(clients, "ecomms_post_delivery_filter", "HEARTBE"), + "unavailable_selected": _find_client(clients, "ecomms_unavailable_selected", "HEARTBE"), + } + before = {name: _delivery(row) for name, row in before_rows.items()} + time.sleep(MEASURE_SECONDS) + sampler_stop.set() + sampler_thread.join(timeout=5.0) + if sampler_thread.is_alive(): + raise RuntimeError("Subscription sampler did not stop") + if sampler_errors: + raise RuntimeError("Subscription sampler failed: {}".format(sampler_errors)) + + clients = broker._fetch_collection("/api/v5/clients") + subscriptions = broker._fetch_collection("/api/v5/subscriptions") + after_rows = { + name: _find_client(clients, row["clientid"].split("_HEARTBE")[0], "HEARTBE") + for name, row in before_rows.items() + } + after = {name: _delivery(row) for name, row in after_rows.items()} + delivery = {name: _delta(before[name], after[name]) for name in before} + topics = _topics_by_client(subscriptions) + + full_id = before_rows["full_network"]["clientid"] + selected_id = before_rows["selected_nodes"]["clientid"] + summary_id = before_rows["summary_discovery"]["clientid"] + negative_id = before_rows["post_delivery_filter"]["clientid"] + unavailable_id = before_rows["unavailable_selected"]["clientid"] + global_ctrl = f"{ROOT_TOPIC}/ctrl" + targeted_ctrl = f"{ROOT_TOPIC}/ctrl/{selected_node}" + unavailable_ctrl = f"{ROOT_TOPIC}/ctrl/{unavailable_node}" + + assert topics.get(full_id) == [global_ctrl], topics.get(full_id) + assert topics.get(selected_id) == [targeted_ctrl], topics.get(selected_id) + assert global_ctrl not in topics.get(selected_id, []) + assert topics.get(summary_id, []) == [], topics.get(summary_id) + assert topics.get(negative_id) == [global_ctrl], topics.get(negative_id) + assert topics.get(unavailable_id) == [unavailable_ctrl], topics.get(unavailable_id) + for sample in subscription_samples: + for reduced_name in [ + "ecomms_selected_nodes", + "ecomms_summary_discovery", + "ecomms_unavailable_selected", + ]: + sampled_topics = _session_topics(sample, reduced_name) + assert global_ctrl not in sampled_topics, { + "session": reduced_name, + "topics": sampled_topics, + } + assert delivery["full_network"]["messages"] > delivery["selected_nodes"]["messages"] > 0 + assert delivery["summary_discovery"]["messages"] == 0 + assert delivery["unavailable_selected"]["messages"] == 0 + assert delivery["post_delivery_filter"]["messages"] > delivery["selected_nodes"]["messages"] + assert delivery["full_network"]["bytes"] > delivery["selected_nodes"]["bytes"] > 0 + + statuses = { + name: session.get_heartbeat_observation_status() + for name, session in sessions.items() + } + assert statuses["selected_nodes"]["state"] == "ready", statuses + assert statuses["selected_nodes"]["accepted_heartbeats"] > 0, statuses + assert statuses["selected_nodes"]["accepted_summaries"] == 0, statuses + assert statuses["selected_nodes"]["last_valid_sender"] == selected_node, statuses + assert statuses["selected_nodes"]["last_valid_age_seconds"] <= MEASURE_SECONDS, statuses + assert statuses["summary_discovery"]["state"] == "ready", statuses + assert statuses["summary_discovery"]["accepted_summaries"] > 0, statuses + assert statuses["summary_discovery"]["last_valid_sender"] == summary_publisher, statuses + assert statuses["summary_discovery"]["last_valid_age_seconds"] <= MEASURE_SECONDS, statuses + assert statuses["unavailable_selected"]["state"] == "degraded", statuses + assert statuses["unavailable_selected"]["accepted_heartbeats"] == 0, statuses + assert statuses["unavailable_selected"]["reason"] == "targeted_heartbeat_timeout", statuses + + print(json.dumps({ + "status": "ok", + "selected_node": selected_node, + "summary_publisher": summary_publisher, + "unavailable_node": unavailable_node, + "delivery": delivery, + "subscription_samples": len(subscription_samples), + "observation_status": statuses, + }, indent=2, default=str)) + return 0 + finally: + sampler_stop.set() + sampler_thread.join(timeout=5.0) + for session in sessions.values(): + session.close(wait_close=True) + + +if __name__ == "__main__": + raise SystemExit(main()) From 4621f380a16ea5c45bb516292fa4a92b5c52f613 Mon Sep 17 00:00:00 2001 From: Cristi Bleotiu Date: Wed, 19 Aug 2026 18:58:13 +0300 Subject: [PATCH 2/4] fix: prove heartbeat mirror persisted config What changed: - remove forced targeted mirror env override from the comms compose testbed - assert persisted comms config enables mirror behavior separately from env precedence Why: - fanout validation should fail if persisted config mirroring is ignored --- docker-compose_comms.yaml | 1 - tests/test_commandcontrol_combined_config.py | 11 +++++++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/docker-compose_comms.yaml b/docker-compose_comms.yaml index de40de75..6c193b93 100644 --- a/docker-compose_comms.yaml +++ b/docker-compose_comms.yaml @@ -27,7 +27,6 @@ x-edge-comms-node: &edge-comms-node EE_MQTT_SUBTOPIC: address EE_MQTT_HEARTBEAT_QOS: "1" EE_MQTT_COMMAND_QOS: "2" - EE_HEARTBEAT_TARGETED_MIRROR_ENABLED: "1" EE_NETMON_ORACLE_ONLY_HEARTBEAT_MODE: "1" EE_ENABLE_NETMON_API_PROBE: "1" # The comms testbed does not run child workloads, so DinD TLS only adds a diff --git a/tests/test_commandcontrol_combined_config.py b/tests/test_commandcontrol_combined_config.py index 148f03e1..9b228682 100644 --- a/tests/test_commandcontrol_combined_config.py +++ b/tests/test_commandcontrol_combined_config.py @@ -66,17 +66,24 @@ def test_rollout_docs_name_seed_and_persisted_config_boundaries(self): self.assertIn(".config_app_comms.json", text) self.assertIn("rollback", text.lower()) - def test_live_testbed_enables_mirror_and_has_broker_fanout_probe(self): + def test_live_testbed_has_broker_fanout_probe(self): compose = (REPO_ROOT / "docker-compose_comms.yaml").read_text() probe = (REPO_ROOT / "tests" / "validate_sdk_heartbeat_fanout.py").read_text() - self.assertIn('EE_HEARTBEAT_TARGETED_MIRROR_ENABLED: "1"', compose) + self.assertNotIn('EE_HEARTBEAT_TARGETED_MIRROR_ENABLED', compose) self.assertIn("/api/v5/subscriptions", probe) self.assertIn("send_msg", probe) self.assertIn("send_oct", probe) self.assertIn("post_delivery_filter", probe) self.assertIn("heartbeat_observation_mode", probe) + def test_comms_env_precedence_is_not_testbed_forced(self): + config = self._load(".config_app_comms.json") + self.assertTrue( + config["COMMUNICATION"]["PARAMS"]["HEARTBEAT_TARGETED_MIRROR_ENABLED"], + ) + + if __name__ == "__main__": unittest.main() From fe303e0f964af475b50609bf9660c655fc5c03ba Mon Sep 17 00:00:00 2001 From: Cristi Bleotiu Date: Thu, 20 Aug 2026 12:58:51 +0300 Subject: [PATCH 3/4] test: prove persisted heartbeat mirroring What changed: - require the fanout probe to validate live persisted mirror configuration - reject hidden mirror environment overrides in the primary test Why: - ensure targeted fanout evidence comes from config_app.txt --- tests/test_commandcontrol_combined_config.py | 14 +++--- tests/validate_sdk_heartbeat_fanout.py | 52 ++++++++++++++++++-- 2 files changed, 54 insertions(+), 12 deletions(-) diff --git a/tests/test_commandcontrol_combined_config.py b/tests/test_commandcontrol_combined_config.py index 9b228682..70b0ca14 100644 --- a/tests/test_commandcontrol_combined_config.py +++ b/tests/test_commandcontrol_combined_config.py @@ -69,21 +69,19 @@ def test_rollout_docs_name_seed_and_persisted_config_boundaries(self): def test_live_testbed_has_broker_fanout_probe(self): compose = (REPO_ROOT / "docker-compose_comms.yaml").read_text() probe = (REPO_ROOT / "tests" / "validate_sdk_heartbeat_fanout.py").read_text() + config = self._load(".config_app_comms.json") self.assertNotIn('EE_HEARTBEAT_TARGETED_MIRROR_ENABLED', compose) + self.assertTrue( + config["COMMUNICATION"]["PARAMS"]["HEARTBEAT_TARGETED_MIRROR_ENABLED"], + ) + self.assertIn("config_app.txt", probe) + self.assertIn("Primary fanout validation must not use", probe) self.assertIn("/api/v5/subscriptions", probe) self.assertIn("send_msg", probe) self.assertIn("send_oct", probe) self.assertIn("post_delivery_filter", probe) self.assertIn("heartbeat_observation_mode", probe) - def test_comms_env_precedence_is_not_testbed_forced(self): - config = self._load(".config_app_comms.json") - self.assertTrue( - config["COMMUNICATION"]["PARAMS"]["HEARTBEAT_TARGETED_MIRROR_ENABLED"], - ) - - - if __name__ == "__main__": unittest.main() diff --git a/tests/validate_sdk_heartbeat_fanout.py b/tests/validate_sdk_heartbeat_fanout.py index 8639810c..80f3dd5d 100644 --- a/tests/validate_sdk_heartbeat_fanout.py +++ b/tests/validate_sdk_heartbeat_fanout.py @@ -1,10 +1,10 @@ #!/usr/bin/env python3 """Prove SDK heartbeat modes by broker delivery, not post-delivery filtering. -Run this only against ``docker-compose_comms.yaml``. The probe discovers one -node and one NetMon publisher, starts real Ratio1 SDK sessions for every mode, -then compares EMQX per-client delivered messages and bytes over one shared -traffic window. +Run this inside one node from ``docker-compose_comms.yaml``. The probe first +proves targeted mirroring comes from the node's persisted configuration, then +discovers one node and one NetMon publisher, starts real Ratio1 SDK sessions +for every mode, and compares EMQX delivery over one shared traffic window. """ import importlib.util @@ -30,6 +30,48 @@ PASSWORD = os.environ.get("ECOMMS_MQTT_PASS", "ecomms") DISCOVERY_SECONDS = float(os.environ.get("ECOMMS_SDK_DISCOVERY_SECONDS", "90")) MEASURE_SECONDS = float(os.environ.get("ECOMMS_SDK_MEASURE_SECONDS", "30")) +MIRROR_ENV_KEY = "EE_HEARTBEAT_TARGETED_MIRROR_ENABLED" +PERSISTED_CONFIG_PATH = pathlib.Path(os.environ.get( + "ECOMMS_PERSISTED_CONFIG_PATH", + "/edge_node/_local_cache/_data/box_configuration/config_app.txt", +)) + + +def _assert_persisted_mirror_source(): + """Prove this live node mirrors from persisted config, not an env override. + + Returns + ------- + str + Path of the effective persisted application configuration. + + Raises + ------ + AssertionError + If an environment override exists or persisted mirroring is not enabled. + """ + if MIRROR_ENV_KEY in os.environ: + raise AssertionError( + "Primary fanout validation must not use {}".format(MIRROR_ENV_KEY) + ) + try: + config = json.loads(PERSISTED_CONFIG_PATH.read_text()) + mirror_enabled = config["COMMUNICATION"]["PARAMS"][ + "HEARTBEAT_TARGETED_MIRROR_ENABLED" + ] + except Exception as exc: + raise AssertionError( + "Cannot read persisted mirror setting from {}".format( + PERSISTED_CONFIG_PATH, + ) + ) from exc + if mirror_enabled is not True: + raise AssertionError( + "Persisted heartbeat targeted mirror must be true, got {!r}".format( + mirror_enabled, + ) + ) + return str(PERSISTED_CONFIG_PATH) def _broker_module(): @@ -192,6 +234,7 @@ def sample(): def main(): + persisted_config_path = _assert_persisted_mirror_source() broker = _broker_module() discovered = _discover_publishers() selected_node = discovered["node"] @@ -310,6 +353,7 @@ def main(): print(json.dumps({ "status": "ok", + "persisted_mirror_config": persisted_config_path, "selected_node": selected_node, "summary_publisher": summary_publisher, "unavailable_node": unavailable_node, From d6c903197e1bd4f78d703e7f533f0d916e944ca3 Mon Sep 17 00:00:00 2001 From: Cristi Bleotiu Date: Thu, 20 Aug 2026 19:19:50 +0300 Subject: [PATCH 4/4] test: enforce commandcontrol config types What changed: - assert persisted heartbeat worker and mirror settings use exact JSON boolean and integer types Why: - prevent truthiness-compatible values from hiding malformed deployment configuration --- tests/test_commandcontrol_combined_config.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/test_commandcontrol_combined_config.py b/tests/test_commandcontrol_combined_config.py index 70b0ca14..09c62f48 100644 --- a/tests/test_commandcontrol_combined_config.py +++ b/tests/test_commandcontrol_combined_config.py @@ -20,13 +20,19 @@ def _assert_regrouped(self, config): def _assert_combined_params(self, config, mirror_enabled): params = config["COMMUNICATION"]["PARAMS"] + self.assertIs(type(params["HEARTBEAT_INGRESS_WORKER_ENABLED"]), bool) self.assertTrue(params["HEARTBEAT_INGRESS_WORKER_ENABLED"]) + self.assertIs(type(params["HEARTBEAT_INGRESS_QUEUE_SIZE"]), int) self.assertEqual(params["HEARTBEAT_INGRESS_QUEUE_SIZE"], 10000) + self.assertIs(type(params["HEARTBEAT_AUTH_WORKERS"]), int) self.assertEqual(params["HEARTBEAT_AUTH_WORKERS"], 4) + self.assertIs(type(params["HEARTBEAT_AUTH_MAX_IN_FLIGHT"]), int) self.assertEqual(params["HEARTBEAT_AUTH_MAX_IN_FLIGHT"], 32) self.assertEqual(params["HEARTBEAT_AUTH_MODE"], "shadow") self.assertIn("{}", params["CTRL_CHANNEL"]["TARGETED_TOPIC"]) + self.assertIs(type(params["CTRL_CHANNEL"]["SUBSCRIBE_TARGETED"]), bool) self.assertFalse(params["CTRL_CHANNEL"]["SUBSCRIBE_TARGETED"]) + self.assertIs(type(params["HEARTBEAT_TARGETED_MIRROR_ENABLED"]), bool) self.assertEqual( params["HEARTBEAT_TARGETED_MIRROR_ENABLED"], mirror_enabled,