Writing a Gateway

A gateway is the interface between the MCS and the physical link to a target: a ground station, a bench RF or CAN interface, an EGSE, a simulator. It knows nothing of the platform. It sends the frames the drivers (or transports) produced and hands back the frames it receives, opaque both ways; everything that depends on the platform is the driver's.

             driver                 gateway                      target
TC_COMMANDS ─encode─► frame ─uplink─► send ──────────────────────► (RF, CAN, …)
                                     │ SENT / SEND_FAILED (ACK 2)
                                     ▼
                              stellar.tc.evt.<target>.<tc>

(RF, CAN, …) ───────────────► receive ─► stellar.tm.raw.<gateway>.<target> ─► driver
                                     ├─► stellar.stream.<target>.<stream>   (payload streams)
                                     └─► stellar.metrics.<gateway>.<target> (throughput, 5 s)
heartbeat: health, link_available ──► stellar.ctl.hb.gateway.<gateway>

A gateway has five duties:

1. Uplink. Take the frames of the links it is bound to, send them, report each outcome (ACK 2).
2. Downlink. Publish every frame received, with its ground reception time.
3. Link state. Say in its heartbeats whether the link is available now.
4. Throughput. Count the bytes each way, per target, and report them.
5. Streams. Publish the segments of payload streams, when the link carries some.

With the SDK, you write the first two and the third; the SDK does the rest.

Registration#

A gateway registers with the kind gateway and a gateway section:

A receive-only gateway has no uplink, a send-only one no downlink. The theoretical rates are used when nothing better is known: the transfer manager paces file transfers with them when neither the pass nor the measured throughput gives a rate.

Binding#

A link names its gateway, or a constraint on tags:

targets:
  flatsat-1:
    links:
      nominal: {driver: platform-v3-ccsds@4, transport: ccsds-tc@1, gateway: bench-rf-1, default: true}
  sat1-fm:
    links:
      nominal: {driver: platform-v3-ccsds@4, transport: ccsds-tc@1, gateway: "any(tag: sband)", default: true}

• Named gateway: the link goes through that instance only.
• Constraint (any(tag: sband), any(tag: sband, tag: eu)): any registered gateway carrying all these tags fits. The reconciler keeps the gateway already bound while it fits, and gives a new link to the fitting gateway serving the fewest links, by name on a tie.
• Health. A gateway reporting degraded, or missing three heartbeats, no longer fits: its links are bound elsewhere when possible.
• Stations. During a pass, the gateway of the default link of the target is the one attached to the station: it fits the link and carries the station name as a tag (gs-a). Up to 10 minutes before the AOS the reconciler checks that it is registered and healthy. See Passes.
• The uplink link type of the gateway must be the output of the transport, or of the driver without transport.

The reconciler delivers the links a gateway serves with the bindings verb. For each link, uplink is stellar.tc.uplink.<gateway>.<target>: the links of a target through one gateway share it.

Uplink and ACK 2#

• Consumer. For each uplink subject, the gateway consumes TC_COMMANDS with the durable consumer uplink_<gateway>_<target>, created if missing with deliver_policy: new, ack_wait: 5 s and max_ack_pending: 1: one frame at a time, so that the frames of a target leave in the order they were encoded, even while a link moves from one gateway to another.
• Frame. The payload is the frame, opaque. Headers: Stellar-Correlation (the telecommand ULID), Stellar-Deadline, Stellar-Config, Nats-Msg-Id (<tc_id>-uplink).
• ACK 2, on stellar.tc.evt.<target>.<tc_id> with Nats-Msg-Id <tc_id>-<STATE>:
  • received after its deadline: not sent, SEND_FAILED with received after its deadline, not sent;
  • sent: SENT (the frame left the gateway; verification is the MCS's);
  • not sent: SEND_FAILED with the reason (link unavailable: out of pass, a hardware error…). The MCS never retries on its own: the step or the operator decides.
• Fragments. Frames carrying Stellar-Fragment: <i>/<n> (and Nats-Msg-Id <tc_id>-uplink-<i>) belong to one telecommand: SENT after the last one only; at the first failure SEND_FAILED (frame <i> of <n>: <reason>), and the next frames of that telecommand are dropped without event.
• Frames of no telecommand. A frame without Stellar-Correlation is a CFDP PDU of a driver: send it like the others, without event; CFDP asks again for what is lost.
• Acknowledge each frame once its event is published, whatever the outcome. A frame not acknowledged within 5 s is delivered again; its event is then de-duplicated.
• Out of pass, fail the frames at once (SEND_FAILED) rather than holding them: deadlines are short, and the executor needs an answer.

With the SDK, all of this is done for you: send(target, frame) returning Ok gives SENT, an error gives SEND_FAILED with its message.

Modes and parameters#

A gateway may need to follow the target: a ground modem at the rate the spacecraft transmits in its current mode, for instance. Every binding carries the current mode of its target and its parameters in that mode (numbers in the unit of the parameter, booleans, enum values), and the reconciler delivers the bindings again at each change of mode. The SDKs hand them to the gateway through the LinkContext of the link:

• With each frame. Gateway::send_with(target, frame, context) in Rust, whose default calls send; in Python, a @gateway.send function with three parameters (target, frame, context). The context is resolved for the environment of the telecommand (header Stellar-Environment): parameters of the link and parameters of the target, overrides applied.
• When a link is bound, and each time its binding changes, such as a new mode: Gateway::configure(context) in Rust (default: nothing), @gateway.configure in Python (called only when the context changed). This context has no environment: the parameters are those of the target without overrides.

context.parameter("rf.tx_baudrate") reads a parameter; the parameter of an instance (tcu[TCU1].anode_voltage) falls back on the value given for every instance (tcu.anode_voltage).

impl Gateway for Modem {
    // … software, capabilities, send

    async fn configure(&self, context: &LinkContext) {
        if let Some(rate) = context.parameter("rf.tx_baudrate").and_then(|v| v.as_f64()) {
            self.set_rate(rate).await;
        }
    }
}

@gateway.configure
async def configure(context):
    await modem.set_rate(context.parameter("rf.tx_baudrate"))

@gateway.send
async def send(target, frame, context):
    await modem.write(frame, rate=context.parameter("rf.tx_baudrate"))

Downlink#

• Raw telemetry. Publish every frame received on stellar.tm.raw.<gateway>.<target>, with Stellar-Ground-Time (its ground reception time, RFC 3339, taken as early as possible) and, for data replayed from on-board storage, Stellar-Delivery: deferred. TM_RAW keeps it 30 days. A gateway serving several targets demultiplexes the frames (spacecraft id, CAN node…) and publishes each under its target.
• Streams. A segment of a payload stream goes to stellar.stream.<target>.<stream_id>, raw, with Stellar-Ground-Time and Stellar-Stream-Seq, a number from 1, consecutive while nothing is lost, so that the MCS counts the gaps. The STREAMS stream archives it; the MCS never decodes it. See Continuous Streams.

In the SDKs, Downlink::publish(target, frame, ground_time, deferred) and Downlink::publish_segment(target, stream, seq, segment, ground_time) do it, and count the throughput.

Link state and health#

Each heartbeat carries link_available (satellite in view, station in service, bench cable up; a gateway without this notion always says true) and the health (degraded with a reason when the gateway cannot work normally: antenna fault, lost connection to its hardware). The MCS acts on link_available:

• the transfer manager asks and writes file chunks only while the link is available, and takes up where it stopped at the next pass;
• in orbit, a manual run is refused when the gateway of the default link of its target reports no link (run::no-link);
• the web console and GET /v1/topology show it for each link.

An error returned by receive marks the gateway degraded.

Throughput#

Every metrics_period (5 s), for each target served in the window, the SDK publishes a ThroughputReport on stellar.metrics.<gateway>.<target> (core NATS, not stored): uplink_bps and downlink_bps (mean over the window), uplink_bytes and downlink_bytes (cumulated since start), window_ms. The transfer manager prefers a measured downlink rate; the web console shows the reports on the page of the gateway (GET /v1/instances/gateway/<gateway>/throughput). The Rust SDK also exposes stellar_gateway_throughput_bps and stellar_gateway_bytes_total (by direction and target) and stellar_gateway_uplink_frames_total (by outcome) on its Prometheus endpoint.

A gateway in Rust and Python#

The smallest Rust gateway (examples/rust/gateways/bench) logs the frames it "sends" and receives a status line of a power supply every second; the Python one (examples/python/gateway.py) carries SCPI lines to an instrument over TCP:

use std::sync::Arc;
use std::time::Duration;

use stellar_common::Timestamp;
use stellar_sdk::contract::{GatewayCapabilities, LinkCapability};
use stellar_sdk::{Downlink, Gateway, Options, run_gateway};
use tokio_util::sync::CancellationToken;

struct Bench;

impl Gateway for Bench {
    fn software(&self) -> (String, String) {
        ("bench-gateway".to_owned(), "0.1.0".to_owned())
    }

    fn capabilities(&self) -> GatewayCapabilities {
        let ethernet = LinkCapability { link_type: "ethernet".to_owned(), rate_bps: 100_000_000 };
        GatewayCapabilities {
            tags: vec!["bench".to_owned()],
            uplink: Some(ethernet.clone()),
            downlink: Some(ethernet),
        }
    }

    async fn send(&self, target: &str, frame: &[u8]) -> Result<(), String> {
        tracing::info!(target, frame = %String::from_utf8_lossy(frame), "sent");
        Ok(())
    }

    async fn receive(&self, downlink: Downlink) -> anyhow::Result<()> {
        let mut ticker = tokio::time::interval(Duration::from_secs(1));
        loop {
            ticker.tick().await;
            let frame = "VOLT 28.0;CURR 0.5;OUTP 1".as_bytes().to_vec().into();
            downlink.publish("psu-lab-2", frame, Timestamp::now(), false).await?;
        }
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let shutdown = CancellationToken::new();
    let stop = shutdown.clone();
    tokio::spawn(async move {
        let _ = tokio::signal::ctrl_c().await;
        stop.cancel();
    });
    run_gateway(Arc::new(Bench), &Options::from_env("lab-eth-1"), None, shutdown).await
}

from stellar_mcs import Downlink, Gateway, Link

gateway = Gateway(
    "lab-eth",
    "1.0.0",
    tags=["bench", "ethernet"],
    uplink=Link("ethernet", 100_000_000),
    downlink=Link("ethernet", 100_000_000),
)


@gateway.send
async def send(target: str, frame: bytes) -> None:
    """Returning means SENT; raising means SEND_FAILED with the message as the reason."""
    await instrument.write(frame)


@gateway.receive
async def receive(downlink: Downlink) -> None:
    """Polls the instrument and hands every answer to the MCS, until the gateway stops."""
    while True:
        await instrument.connect()
        try:
            while True:
                await instrument.write(b"STAT?")
                await downlink.publish(args.target, await instrument.readline())
                await asyncio.sleep(args.period)
        except ConnectionError as error:
            log.warning("%s", error)


@gateway.link_available
def available() -> bool:
    return instrument.connected


@gateway.rpc("identify")
async def identify(request: dict[str, object]) -> dict[str, object]:
    """A verb of this gateway: stellar.ctl.rpc.gateway.<instance>.identify."""
    return {"instrument": f"{instrument.host}:{instrument.port}", "connected": available()}


if __name__ == "__main__":
    gateway.run()

In Python, downlink.publish(target, frame, ground_time=None, deferred=False) stamps the ground time now by default, and downlink.publish_segment(target, stream, seq, segment, ground_time=None) publishes a stream segment. There is no gateway_main in Rust: build the Options with Options::from_env(default_instance) and call run_gateway, as above.

Control verbs#

Besides status, credentials, bindings and reregister, a gateway may answer verbs of its own on stellar.ctl.rpc.gateway.<gateway>.<verb>, JSON in and out, {"error": "…"} on failure. In Rust, implement stellar_sdk::Rpc (call(verb, request)) and pass it to run_gateway. The simulated gateway answers faults, fault, files, generate, rewrite, corrupt and stream_gap, which the API (/v1/sim/…) and the tests use.

Permissions#

Once bound, a gateway receives a JWT limited to:

It cannot publish the telemetry of a target it is not bound to, nor under another gateway's name.

The example gateways#

Running a gateway#

Same environment as a driver: STELLAR_NATS_URL, STELLAR_INSTANCE, STELLAR_NATS_CREDENTIALS, STELLAR_NATS_CA, STELLAR_NATS_CERT, STELLAR_NATS_KEY, and in Rust STELLAR_METRICS_ADDR. See Writing a Driver.

STELLAR_INSTANCE=bench-can-1 bench-can --interface can0 --target flatsat-1
STELLAR_INSTANCE=lab-eth-1 uv run gateway.py --target psu-lab-2 --instrument 127.0.0.1:5025

Keep gateways synchronised with NTP or PTP: their ground reception time is the reference of freshness and of within windows in the whole chain.