rmon/docs/protocol.md

RMon Wire Protocol

Transport

• TCP over an SSH local port-forward tunnel
• Port: configurable, default 7891
• One persistent TCP connection per UI client
• Multiple clients can connect to the same agent simultaneously

Framing

Every message (in both directions) uses the same length-prefix frame:

┌─────────────────────┬────────────────────────────┐
│  length : u32 LE    │  payload : [u8; length]    │
└─────────────────────┴────────────────────────────┘

• length is the byte count of the serialized payload (excludes the 4-byte length field itself)
• payload is a bincode-serialized ClientMessage (client→agent) or AgentMessage (agent→client)
• bincode configuration: little-endian, fixed-size integers, no trailing bytes

Maximum frame size: 16 MiB (length values above this are a protocol error; close connection).

Versioning

The Handshake / HandshakeAck exchange negotiates compatibility. If the agent does not support the client's version it sends HandshakeAck { accepted: false } and closes the connection. Both sides must be built from the same rmon-common version to guarantee bincode schema compatibility.

Client → Agent Messages

enum ClientMessage {
    /// Must be the first message sent. No other messages are valid before HandshakeAck.
    Handshake {
        protocol_version: u16,   // current: 1
        client_version:   String,
    },

    /// Request the full list of signals the agent knows about.
    /// Agent replies with one SignalList message.
    ListSignals,

    /// Subscribe to live data for the given signals.
    /// Agent begins sending DataBatch messages for these signals.
    /// `from_ns`: if Some, also request buffered/stored data from that timestamp forward
    ///            before live data begins.
    Subscribe {
        ids:     Vec<SignalId>,
        from_ns: Option<i64>,
    },

    /// Stop live delivery for these signals. Does not affect storage.
    Unsubscribe {
        ids: Vec<SignalId>,
    },

    /// Request stored historical data for a signal.
    /// Agent replies with one or more HistoryChunk messages (done=true on last chunk).
    /// `max_points`: agent may downsample to fit; 0 = no limit.
    QueryHistory {
        id:         SignalId,
        from_ns:    i64,
        to_ns:      i64,
        max_points: u32,
    },

    /// Start recording the given signals. Mode overrides any existing mode for these signals.
    StartRecording {
        ids:  Vec<SignalId>,
        mode: StorageMode,
    },

    /// Stop recording (data already written is kept).
    StopRecording {
        ids: Vec<SignalId>,
    },

    /// Request current agent status (version, uptime, source states).
    GetStatus,
}

Agent → Client Messages

enum AgentMessage {
    /// Response to Handshake.
    HandshakeAck {
        protocol_version: u16,
        agent_version:    String,
        accepted:         bool,
        reject_reason:    Option<String>,
    },

    /// Full signal list, sent in response to ListSignals.
    /// Sent unsolicited again if the signal list changes (source added/removed).
    SignalList {
        signals: Vec<SignalInfo>,
    },

    /// Live data batch for subscribed signals.
    /// The agent batches samples collected since the last send (target: ~50ms intervals).
    /// Multiple signals may be grouped into one message to reduce framing overhead.
    DataBatch {
        batches: Vec<SignalBatch>,
    },

    /// One chunk of a historical query response.
    HistoryChunk {
        id:      SignalId,
        samples: Vec<Sample>,
        done:    bool,   // true on the final chunk for this query
    },

    /// Acknowledgement of StartRecording / StopRecording.
    RecordingStatus {
        id:        SignalId,
        recording: bool,
        mode:      Option<StorageMode>,
    },

    /// Response to GetStatus.
    StatusReport {
        agent_version: String,
        hostname:      String,
        uptime_secs:   u64,
        sources:       Vec<SourceStatus>,
    },

    /// Unrecoverable error for a specific operation. Connection remains open.
    Error {
        context: String,   // e.g. "QueryHistory" or "Subscribe"
        message: String,
    },
}

struct SignalBatch {
    id:      SignalId,
    samples: Vec<Sample>,
}

struct SourceStatus {
    name:      String,
    source_type: SourceType,
    connected: bool,
    last_sample_ns: Option<i64>,
    error:     Option<String>,
}

Session Lifecycle

Client                                  Agent
  │                                       │
  │──── Handshake ────────────────────►   │
  │                                       │
  │◄─── HandshakeAck (accepted=true) ───  │
  │                                       │
  │──── ListSignals ───────────────────►  │
  │◄─── SignalList ─────────────────────  │
  │                                       │
  │──── Subscribe {ids, from_ns} ──────►  │  ←── live data starts flowing
  │◄═══ DataBatch (repeated) ═══════════  │
  │                                       │
  │──── QueryHistory ──────────────────►  │
  │◄─── HistoryChunk (done=false) ──────  │
  │◄─── HistoryChunk (done=false) ──────  │
  │◄─── HistoryChunk (done=true) ───────  │
  │                                       │
  │──── StartRecording ────────────────►  │
  │◄─── RecordingStatus ────────────────  │
  │                                       │
  │  [TCP close / SSH tunnel drops]       │
  │                                       │  ←── agent continues running,
                                               data continues to be stored

Batching Policy

The agent collects samples from all sources in a shared tokio::broadcast channel. A per-session task drains the channel and sends DataBatch messages at a target interval of 50 ms (20 Hz UI update rate). If a batch would exceed 1 MiB, it is split and sent immediately.

High-frequency signals (e.g., 10 kHz UDP) are automatically decimated to 10× the batch rate (200 Hz) for live display; full-resolution data is still stored.

Error Handling

• Error messages are non-fatal; the session continues
• If the agent cannot fulfill a QueryHistory (signal not found, time range not stored), it sends Error then continues
• If the TCP connection drops, the agent discards the session's subscriptions; no other sessions are affected
• The UI reconnects automatically (with exponential backoff) if the tunnel drops