feat: configure staged CommandControl rollout

.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"
             },

.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
-}
+}

.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

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"
       },

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.

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/<node_address>` 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.

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"
+  }
+}

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
+}