Skip to content
Brokkr

brokkr-agent::broker_ws

Rust

Agent side of the internal broker↔agent WebSocket channel.

Wire protocol: brokkr_wire. Broker side: brokkr_broker::ws.

This module owns:

  • the dial-and-reconnect loop against /internal/ws/agent,
  • a public WsState watch channel so other agent components (heartbeat, event emitter, work-order receiver) can choose between WS and REST in real time,
  • the per-direction message queues that those components feed and drain.

Per ADR-0008 the WS channel is on by default; REST polling is the automatic fallback. Setting agent.ws_force_rest = true (env or config) short-circuits the whole thing — no dial attempt is ever made and the state stays WsState::ForceRestOnly for the lifetime of the agent.

Structs

Section titled “Structs”

struct WsClient

Section titled “struct WsClient”

pub

Public handle to the WS client. Constructed by spawn; the connection task runs in the background until the returned JoinHandle is dropped or the agent process exits.

Methods

Section titled “Methods”
state
Section titled “state”

pub

fn state(&self) -> watch::Receiver<WsState>

Watch the connection state. Cheap to clone, safe to await.

Source 
    pub fn state(&self) -> watch::Receiver<WsState> {
        self.state.clone()
    }
outbound
Section titled “outbound”

pub

fn outbound(&self) -> mpsc::Sender<WsMessage>

Sender for outbound messages (heartbeat, agent events, health, streamed kube events, log lines). If the lane is full or WS is down the send returns Err; callers should fall back to REST.

Source 
    pub fn outbound(&self) -> mpsc::Sender<WsMessage> {
        self.outbound_tx.clone()
    }
Section titled “uplink”

pub

fn uplink(&self) -> WsUplink

Cheap clonable handle bundling the outbound sender with a current state view, intended for emitter call sites that want a single “try WS, else REST” decision point (see WsUplink::try_send).

Source 
    pub fn uplink(&self) -> WsUplink {
        WsUplink {
            state: self.state.clone(),
            outbound: self.outbound_tx.clone(),
        }
    }
take_​inbound
Section titled “take_​inbound”

pub

fn take_inbound(&mut self) -> Option<mpsc::Receiver<WsMessage>>

Take ownership of the inbound receiver. Single consumer; subsequent calls return None.

Source 
    pub fn take_inbound(&mut self) -> Option<mpsc::Receiver<WsMessage>> {
        self.inbound_rx.take()
    }
Section titled “struct WsUplink”

pub

Derives: Clone

Send-side handle for agent components that want to prefer WS but fall back to REST when WS is down (per ADR-0008). Cheap to Clone — share one per emitter rather than locking the WsClient.

Methods

Section titled “Methods”
is_​up
Section titled “is_​up”

pub

fn is_up(&self) -> bool

True iff the WS state is currently Up. ForceRestOnly and Down both return false, so call sites short-circuit cleanly when the agent is configured for REST-only.

Source 
    pub fn is_up(&self) -> bool {
        matches!(*self.state.borrow(), WsState::Up)
    }
try_​send
Section titled “try_​send”

pub

fn try_send(&self, msg: WsMessage) -> Result<(), WsMessage>

Try to send a message over WS. Returns Err(msg) (giving the caller their message back) if WS is down or the outbound lane is full or closed — that signals the caller to take the REST path.

Source 
    pub fn try_send(&self, msg: WsMessage) -> Result<(), WsMessage> {
        if !self.is_up() {
            return Err(msg);
        }
        match self.outbound.try_send(msg) {
            Ok(()) => Ok(()),
            Err(tokio::sync::mpsc::error::TrySendError::Full(m))
            | Err(tokio::sync::mpsc::error::TrySendError::Closed(m)) => Err(m),
        }
    }

Enums

Section titled “Enums”

enum WsState

Section titled “enum WsState”

pub

Current state of the WS channel from the agent’s point of view.

Variants

Section titled “Variants”
  • Down - Not connected. Other components must use REST.
  • Up - Connected to the broker. Other components should prefer WS.
  • ForceRestOnly - ws_force_rest = true — the agent will never attempt to use WS. Distinct from Down so callers can short-circuit “wait for WS” logic and avoid logging spurious “WS still down” warnings.
  • AuthRejected - The broker rejected the agent’s PAK on the WS upgrade repeatedly (HTTP 401/403). Terminal: the reconnect loop has given up because the credential is almost certainly revoked (a rotation/delete — see [[BROKKR-T-0176]]). Recovery requires a new PAK + agent restart, so we stop hammering a handshake that will never succeed. Distinct from Down so callers can surface “credential dead” rather than “WS flapping”.

Methods

Section titled “Methods”
is_​up
Section titled “is_​up”

pub

fn is_up(self) -> bool
Source 
    pub fn is_up(self) -> bool {
        matches!(self, WsState::Up)
    }

Functions

Section titled “Functions”

fn spawn

Section titled “fn spawn”

pub

fn spawn(settings: &Settings) -> WsClient

Spawn the WS connection task and return a client handle.

When settings.agent.ws_force_rest == true, no task is spawned beyond a no-op keep-alive that holds the watch sender; the state never leaves ForceRestOnly. Outbound sends fail closed so callers reliably fall back to REST.

Source 
pub fn spawn(settings: &Settings) -> WsClient {
    if settings.agent.ws_force_rest {
        info!("ws_force_rest=true; WebSocket channel disabled, REST-only mode");
        let (state_tx, state_rx) = watch::channel(WsState::ForceRestOnly);
        let (outbound_tx, mut outbound_rx) = mpsc::channel::<WsMessage>(1);
        let (inbound_tx, inbound_rx) = mpsc::channel::<WsMessage>(1);
        // Hold the senders so the watch isn't poisoned and inbound rx
        // doesn't observe an EOF, but never produce any messages.
        let task = tokio::spawn(async move {
            let _state_tx = state_tx;
            let _inbound_tx = inbound_tx;
            // Drain outbound so callers' `try_send` doesn't fill up and
            // give them an illusion of WS-up; we just discard.
            while let Some(msg) = outbound_rx.recv().await {
                debug!(?msg, "discarding WS outbound message: ws_force_rest=true");
            }
        });
        return WsClient {
            state: state_rx,
            outbound_tx,
            inbound_rx: Some(inbound_rx),
            _task: task,
        };
    }


    let (state_tx, state_rx) = watch::channel(WsState::Down);
    let (outbound_tx, outbound_rx) = mpsc::channel::<WsMessage>(OUTBOUND_CAPACITY);
    let (inbound_tx, inbound_rx) = mpsc::channel::<WsMessage>(INBOUND_CAPACITY);


    // Honor ws_url override when set (e.g. WS through a separate ingress, or
    // through the I-0020 A2 chaos toxiproxy). Default: derive from broker_url.
    let url = settings
        .agent
        .ws_url
        .as_deref()
        .map(|s| s.to_string())
        .unwrap_or_else(|| ws_url_from_broker_url(&settings.agent.broker_url));
    let pak = settings.agent.pak.clone();
    let task = tokio::spawn(reconnect_loop(url, pak, state_tx, inbound_tx, outbound_rx));


    WsClient {
        state: state_rx,
        outbound_tx,
        inbound_rx: Some(inbound_rx),
        _task: task,
    }
}

fn ws_​url_​from_​broker_​url

Section titled “fn ws_​url_​from_​broker_​url”

pub

fn ws_url_from_broker_url(broker_url: &str) -> String

Convert http(s)://broker/api/v1-style URLs into the ws(s)://broker/internal/ws/agent form the broker exposes. The agent’s broker_url setting points at the broker root (without /api/v1), so we just swap the scheme and append the path.

Source 
pub fn ws_url_from_broker_url(broker_url: &str) -> String {
    let trimmed = broker_url.trim_end_matches('/');
    let with_ws_scheme = if let Some(rest) = trimmed.strip_prefix("https://") {
        format!("wss://{rest}")
    } else if let Some(rest) = trimmed.strip_prefix("http://") {
        format!("ws://{rest}")
    } else {
        // Unrecognised scheme — assume ws:// and let the connect fail
        // loudly rather than silently mangling the URL.
        format!("ws://{trimmed}")
    };
    format!("{with_ws_scheme}/internal/ws/agent")
}