Skip to main content

Robust streaming

Live data is delivered over server-streaming RPCs, and maritime links are low-bandwidth, high-latency, and intermittent. Design streams to stay lean and to recover automatically.

Treat a stream as resumable​

Each stream sends the current value first, then updates. If a stream ends — cleanly or with an error — reopen it. On reconnect you again receive the current value, so your view converges without special handling.

Reconnect with backoff​

Don't reconnect in a tight loop. Use exponential backoff with jitter and a cap (e.g. start at 500 ms, double to a 30 s ceiling). Reset the backoff once a stream has been healthy for a while.

reconnect.py
import time, random, grpc
from greenroom.vessel.v1 import vessel_service_pb2, vessel_service_pb2_grpc

def stream_vessel_state(channel, vessel_id):
    client = vessel_service_pb2_grpc.VesselStateServiceStub(channel)
    backoff = 0.5
    while True:
        try:
            request = vessel_service_pb2.StreamVesselStateRequest(vessel_id=vessel_id)
            for response in client.StreamVesselState(request):
                backoff = 0.5  # healthy — reset
                yield response.state
        except grpc.RpcError as err:
            time.sleep(backoff + random.uniform(0, backoff))
            backoff = min(backoff * 2, 30.0)
Distinguish a clean end from an error

Not every status code warrants the same response. Retry transient codes (UNAVAILABLE, DEADLINE_EXCEEDED); don't hammer on terminal ones. See Error handling.

Run streams concurrently​

Subscribe to each domain you need on its own stream (vessel, perception, mission, …) and process them concurrently — async tasks, threads, or your runtime's equivalent. Keep per-message work light; offload heavy processing so you don't stall the stream.

Subscribe to only what you need​

Open streams for the domains you actually use. Where a stream offers a choice (for example, filtered vs. unfiltered perception tracks), prefer the narrower one. Less data on the wire means lower latency and fewer drops.

Tolerate gaps​

Assume any stream can pause and resume. Render the last known value with its timestamp and surface staleness in your UI rather than blocking. Use the header.stamp on messages to reason about age.

Keep connections warm​

Configure gRPC keepalive so idle connections are detected promptly and dead ones are torn down, which makes reconnection responsive. Reuse a single channel across all stubs rather than opening one per call.

channel.py
options = [
    ("grpc.keepalive_time_ms", 20000),          # ping every 20s
    ("grpc.keepalive_timeout_ms", 10000),       # fail if no ack in 10s
    ("grpc.keepalive_permit_without_calls", 1), # ping even when idle
]
channel = grpc.secure_channel(endpoint, channel_creds, options=options)
Tune to the link, and agree with the server

Aggressive keepalive on a high-latency satellite link can do more harm than good. Pick intervals that match your link, and confirm them against the endpoint's server-side keepalive policy.

Build for air-gapped deployment​

If your build or runtime environment cannot reach the Buf Schema Registry, vendor the generated code: generate it once with buf generate buf.build/greenroom-robotics/platform-sdk (see Install) and commit the output, pinned to a schema version.

What's next​