Fleet Telemetry#

This worked example connects edge publishers to an operations fold through a regional gateway. It uses hierarchical channels, subnet policy, and permission tokens to show where routing and authority decisions belong. The snippets describe the three roles and their configuration. A production fleet also needs provisioning, key rotation, route setup, capacity planning, monitoring, and a tested failure model.

The shape#

┌─────────────────────────────────────────────┐
│       Operations cluster (region: ops)      │
│  ┌──────────────────┐ ┌──────────────────┐  │
│  │ Telemetry fold   │ │ Dashboard reader │  │
│  │ (CortEX adapter) │ │ (watcher stream) │  │
│  └─────────┬────────┘ └─────────┬────────┘  │
└────────────┼─────────────────────┼──────────┘
             │                     │
        ┌────┴─────────────────────┴────┐
        │  Regional gateway (subnet 3)  │
        │  Channel: telemetry/*         │
        │  Visibility: Exported → ops   │
        └────────────────┬──────────────┘
                         │
         ┌───────────────┼───────────────┐
         ▼               ▼               ▼
    ┌────────┐      ┌────────┐      ┌────────┐
    │ veh-1  │      │ veh-2  │      │ veh-3  │
    │ subnet │      │ subnet │      │ subnet │
    │ 3.7.1  │      │ 3.7.2  │      │ 3.7.3  │
    └────────┘      └────────┘      └────────┘

Three concepts do the work. Channels with hierarchical names carry events upward. Subnets keep each vehicle's internal traffic isolated. A CortEX fold materializes the aggregate view the operations cluster queries.

Setting up the channel hierarchy#

Each vehicle publishes to a channel rooted at its identity:

• vehicles/v-001/telemetry/imu
• vehicles/v-001/telemetry/gps
• vehicles/v-001/telemetry/battery

Internal vehicle channels (e.g. vehicles/v-001/internal/diagnostics) are configured SubnetLocal and never leave the vehicle. The telemetry channels are configured Exported, with the operations cluster's subnet (region: ops, SubnetId::new(&[0])) as the destination:

use net::adapter::net::channel::{ChannelConfig, ChannelName, Visibility};
use net::adapter::net::behavior::capability::CapabilityFilter;
 
// The fleet's root of trust: the entity whose signature every vehicle's and
// operator's token chain must ultimately root at. In a real deployment this
// is a long-lived offline key, distributed to nodes as a public EntityId.
let fleet_root_entity_id = fleet_root_keypair.entity_id().clone();
 
let channel = ChannelName::new("vehicles/v-001/telemetry/imu")?;
let telemetry_cfg = ChannelConfig::new(channel.id())
    .with_visibility(Visibility::Exported)
    .with_publish_caps(CapabilityFilter::new().require_tag("role.vehicle"))
    .with_subscribe_caps(
        CapabilityFilter::new()
            .require_tag("role.operator")
            .require_tag("tier.production"),
    )
    // Anchors the root of trust AND turns on token enforcement.
    .with_token_roots(vec![fleet_root_entity_id])
    .with_priority(4)
    .with_reliable(false)
    .with_rate_limit(100);
 
mesh.register_channel(telemetry_cfg);   // synchronous; no await

The capability filters are routing, not enforcement. publish_caps and subscribe_caps match a node's self-advertised capabilities, so a peer that wants past them can just advertise role.vehicle — they express which nodes this channel is for, not which nodes are allowed (Channels).

What actually restricts the channel is the token gate. Each vehicle holds a token, rooted at the fleet root, scoped to its own telemetry channels; the operations cluster holds tokens scoped to the export destinations. A presented chain must root at fleet_root_entity_id, bind at its leaf to the presenting entity, and authorize the action at every link.

Don't call with_require_token(true) without also setting token_roots. That combination is a valid fail-closed state — and therefore not an error — but with no root to anchor against, nothing can satisfy the gate and every publish and subscribe is denied. It's almost always a typo for with_token_roots(...), which sets require_token for you. The substrate logs a warning at channel registration when it sees this; if telemetry goes silent right after you add auth, check for it.

Configuring subnets#

Each vehicle is its own subnet at the third level of the hierarchy. The four-level subnet ID maps to (region, fleet, vehicle, subsystem):

use std::collections::HashMap;
use std::sync::Arc;
use net::adapter::net::subnet::{SubnetPolicy, SubnetRule};
 
// Each rule maps a tag prefix to one hierarchy level and each tag value to
// that level's byte; the rules combine to build (region, fleet, vehicle).
let vehicle_policy = SubnetPolicy::new()
    .add_rule(SubnetRule {
        tag_prefix: "region:".into(),
        level: 0,
        values: HashMap::from([("west".into(), 3u8)]),   // region byte 3
    })
    .add_rule(SubnetRule {
        tag_prefix: "fleet:".into(),
        level: 1,
        values: HashMap::from([("alpha".into(), 7u8)]),  // fleet byte 7
    })
    .add_rule(SubnetRule {
        tag_prefix: "vehicle:".into(),
        level: 2,
        values: HashMap::from([("v-001".into(), 1u8)]),  // vehicle byte 1
    });
 
// The policy is supplied at mesh construction, not via a runtime setter:
//   Mesh::builder(bind_addr, &psk).with_subnet_policy(Arc::new(vehicle_policy))

The fleet's regional gateway sits at SubnetId::new(&[3, 7]) and handles the export rules — telemetry channels marked Exported ride the export table to the operations subnet (SubnetId::new(&[0])), and everything else stops at the gateway.

The gateway's enforcement is header-only: it reads subnet_id and channel_hash from the packet header, looks up the channel's Visibility, and forwards or drops accordingly. No payload decryption, no per-flow state, no opportunity for an internal channel to leak.

The vehicle's publisher#

Each vehicle runs a small process that reads its sensors and publishes:

use net::{EventBus, EventBusConfig, Event, AdapterConfig};
use std::time::Duration;
 
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = EventBusConfig::builder()
        .adapter(AdapterConfig::net()
            .listen("0.0.0.0:7777")
            .peer("gateway.fleet-west.local:7777"))
        .build()?;
 
    let bus = EventBus::new(config).await?;
 
    let mut interval = tokio::time::interval(Duration::from_millis(10));
    let mut sensors = open_sensors();
 
    loop {
        interval.tick().await;
        let reading = sensors.read();
        let event = Event::from_str(&serde_json::to_string(&reading)?)?;
        bus.ingest(event)?;
    }
}

The vehicle ingests onto its local bus; the NetAdapter ships events through the mesh to anyone subscribed. The gateway picks them up because it's the next-hop forwarder; subscribers in the operations cluster pick them up because they're the destination subnet.

The operations fold#

The operations cluster materializes a per-vehicle view from the incoming telemetry. A CortEX fold consumes the events and updates an aggregate state:

use net::adapter::net::cortex::{CortexAdapter, RedexFold, FoldError};
use net::adapter::net::state::CausalEvent;
use std::collections::HashMap;
 
#[derive(Default, Clone, serde::Serialize, serde::Deserialize)]
struct FleetState {
    vehicles: HashMap<String, VehicleSummary>,
}
 
#[derive(Default, Clone, serde::Serialize, serde::Deserialize)]
struct VehicleSummary {
    last_imu:     ImuReading,
    last_gps:     GpsReading,
    battery_pct:  u8,
    health_score: f32,
    updated_at:   u64,
}
 
struct FleetFold;
 
impl RedexFold<FleetState> for FleetFold {
    fn apply(&self, state: &mut FleetState, event: &CausalEvent)
        -> Result<(), FoldError> {
        let parsed: VehicleEvent = serde_json::from_slice(&event.payload)
            .map_err(|e| FoldError::InvalidPayload(e.to_string()))?;
        let summary = state.vehicles.entry(parsed.vehicle_id).or_default();
        match parsed.kind {
            VehicleEventKind::Imu(r)     => summary.last_imu = r,
            VehicleEventKind::Gps(r)     => summary.last_gps = r,
            VehicleEventKind::Battery(p) => summary.battery_pct = p,
        }
        summary.updated_at = parsed.timestamp_ms;
        summary.health_score = compute_health(summary);
        Ok(())
    }
}

Open a CortEX adapter against the channel that catches all incoming telemetry (a RedEX log subscribed to vehicles/*/telemetry/*):

let fleet_adapter = CortexAdapter::open(
    &redex,
    "fleet-summary",
    operator_origin_hash,
    FleetFold,
).await?;

The fold task subscribes to the RedEX tail, applies events as they arrive, and persists the resulting state. Any operator-side reader can query the state:

let snapshot = fleet_adapter.state().read();
for (vehicle_id, summary) in &snapshot.vehicles {
    println!("{}: battery {}%, health {:.2}",
             vehicle_id, summary.battery_pct, summary.health_score);
}

The dashboard#

The operator's dashboard wants live updates, not polling. The CortEX watcher API gives them deltas as the fold updates:

use futures::StreamExt;
 
let mut stream = Box::pin(
    fleet_adapter
        .watch()
        .stream(),
);
 
while let Some(state) = stream.next().await {
    dashboard.render(&state);
}

The watcher emits the current state on subscribe, then dedupe-emits on every state change. The dashboard renders deterministically; the operator sees telemetry update in close to real time without writing a polling loop.

Putting it together#

The deployment has three roles, each with a small responsibility:

• Vehicles run a publisher process that reads sensors and ingests events. They're in their own subnet, they hold per-vehicle permission tokens, and their SubnetLocal channels never leave the vehicle's mesh.
• Gateways are configured with the export table for the fleet — they know which Exported channels can travel to which destination subnets, and they enforce the rules at packet-header speed.
• Operations runs the fold, the watchers, and the dashboard. It subscribes to telemetry on the export, materializes per-vehicle state through the fold, and exposes live views to operators.

The configured topology routes exported traffic through the gateway. Permission tokens rooted at fleet_root_entity_id enforce publish and subscribe authority; the capability filters remain advisory routing predicates. The operations fold receives events that reach its subscribed channel pattern and survive the configured transport, authorization, and retention boundaries.

Adding a new vehicle#

The operational cost of adding a vehicle to the fleet is one provisioning step: give the new node a keypair, an EntityKeypair, a capability set including role.vehicle and fleet.west, and a token scoped to vehicles/v-NNN/telemetry/*. The subnet policy picks it up automatically (the rule matches on vehicle and fleet.west), the gateway picks up the new export automatically (channels marked Exported flow through to the operations subnet), and the operations fold picks up the new events automatically (it subscribes to the pattern, not to individual channels).

The capability advertisement lets matching rules discover the new node. Provisioning still supplies its identity, token, subnet policy, peer connectivity, and operator records.

What this gives you#

For this example, each vehicle publishes through a gateway and the operations cluster maintains the fold. Actual process placement, traffic, isolation, and read capacity depend on event size, rate limits, gateway topology, retention, and failure policy.

Larger fleets can partition gateways and folds by region or fleet, but the required capacity and failure behavior must be measured for that deployment. The hierarchy organizes the topology; it does not remove the need to size and operate it.