Contracts Service

Implementation reference for the contracts-service. For its architectural role, see Contracts Service Architecture; for the domain model, see Telemetry Distribution.

The only Rust service in the platform. Built on Axum and Tokio, it runs an HTTP API and a background telemetry tunnel in a single process, holding per-building preferences in memory for a lock-free hot path.

Source Layout

The source is organised by role: a composition root and shared kernel at the crate root, the two inbound adapters (api/ for HTTP, tunnel.rs for the broker stream), and outbound integrations plus cross-cutting infrastructure grouped under infra/.

graph TD
    MAIN["main.rs\ncomposition root"] --> API["api/\ndashboard.rs, data.rs, init.rs"]
    MAIN --> TUNNEL["tunnel.rs\nbackground worker"]
    MAIN --> DB["infra/db.rs\nconnect + hydrate"]
    API --> STATE["state.rs\nAppState: DashMap + Mongo handle"]
    TUNNEL --> STATE
    API --> DISC["infra/discovery.rs"]
    API --> METRICS["infra/metrics.rs"]
    TUNNEL --> METRICS
    STATE -.-> DB

Configuration

State and Data Model

pub struct AppState {
    pub building_preferences: Arc<DashMap<String, Vec<String>>>, // hot path
    pub mongo_col: Collection<PreferenceDocument>,               // durable
}

pub struct PreferenceDocument {
    pub id: Option<ObjectId>,
    pub building_id: String,      // unique index
    pub allowed_columns: Vec<String>,
}

DashMap shards its internal locks across cores, so the tunnel’s millions of reads almost never contend with an occasional write. The Mongo collection enforces one document per building via a unique index on building_id.

Startup Sequence

1. connect(MONGO_URI)          -> ensure unique index on building_id
2. load_all()                  -> seed the DashMap from persisted documents
3. start_telemetry_tunnel()    -> spawn the background filter task
4. axum::serve(router)         -> begin handling HTTP

Hydration at boot means a restart restores all preferences before the first request is served.

Persistence Model (Hot Path / Cold Path)

Reads and the tunnel operate purely in memory; durability is handled off the request path.

• Hot path (reads, tunnel): O(1) lock-free DashMap lookups; the tunnel never blocks on a write.
• Cold path (writes): update_preferences mutates the DashMap immediately, then spawns a fire-and-forget Tokio task to upsert_preference into Mongo. The HTTP response returns 200 without waiting for the database. The same write-through applies to init when it creates a new entry.

HTTP API

All routes except /health and /metrics require the mesh-verified x-gateway-claims header (decoded by a GatewayClaims extractor that returns 401 on a missing or malformed header) — the same contract every other service consumes. The user-facing routes are gated at the edge (require_gateway_auth in the Caddyfile, the /contracts AuthorizationPolicy rule in Kubernetes); POST /preferences/init is an internal call and receives the building creator’s claims forwarded by twin-service. This enforces authentication only; per-building authorization is deferred to the Cedar phase.

The Telemetry Tunnel

Started at boot, the tunnel opens a dedicated Redis PubSub connection and subscribes to telemetry:raw.

sequenceDiagram
    participant R as Redis (telemetry:raw)
    participant T as Tunnel
    participant D as DashMap
    participant F as Redis (telemetry:filtered:{id})
    R-->>T: { type, buildingId, roomId, value, ... }
    Note over T: spawn a task per message (ingestion never blocks)
    T->>D: is buildingId a known building?
    alt known
        T->>F: PUBLISH telemetry:filtered:{building_id} (full payload)
    end

• Routing keys on the event’s own buildingId: a reading is published to that one building’s channel and no other, so it can never leak into another tenant’s feed. The payload is forwarded intact (no field stripping).
• There is no metric-type gate. Every metric a known building emits is forwarded; the building’s allowed_columns only drive what the dashboard table displays (client-side). The DashMap is consulted just to confirm the building is known.
• Each message is processed in its own task::spawn, so processing time never back-pressures the subscription stream.
• The routing decision is a pure function (resolve_channel) so it is unit-tested in isolation; the worker increments received/published counters and observes fan-out latency, exported via GET /metrics.

Metric Discovery

GET / builds the catalog the frontend uses to populate column pickers.

1. discover_services() collects every environment value whose key ends in _METRICS_URL.
2. Each service’s GET /contracts is fetched and deserialised (either a ServiceMetricsContract or a bare metric list).
3. Each metric is tagged with its source_service and added via push_unique_metric, which deduplicates on (metric_key, interface_name, source_service).

New metric types therefore appear automatically once a service advertises them; no change is needed here.