Streaming: SSE, Documents, Chunked, Bytestream, and Resumable Downloads

A streaming endpoint isn’t special machinery — it’s an ordinary endpoint whose Output is a stream. Set three things and you have one: a StreamIterator Output, a Decoder for the framing, and a companion trait that decodes one frame.

The foundation: one StreamIterator

Every stream handle in Capi is a StreamIterator<T, Meta = ()>. Its item type is hardwired to Result<T, QueryError> (the second generic is per-stream metadata, not the error), and it exposes inherent next() / try_next() / try_collect() / try_for_each() / metadata() / size_hint() plus the adapters map, try_map, filter, filter_map, take, take_while, skip, skip_while, inspect, count, and nth. It also implements futures::Stream (async builds) or Iterator (sync builds) — one or the other, selected by feature, never both at once — so it drops into the combinators you already use.

That one type is also what every framing produces — the Decoder and its companion trait are what differ, not the handle. Pick a framing by pairing a decoder marker with the trait you implement:

Two decoders do name an alias, because their handle carries a contract of its own: FlatMapIterator<T> and ResumableIterator<T> are StreamIterator<T> with the metadata slot closed, and their items stay QueryError-fallible because those streams issue follow-up requests. Paginated endpoints likewise use PagedIterator<P>.

A worked SSE stream

grok’s Chat Completions stream is an OverrideEndpoint over the buffered chat request: it swaps the decoder, asks for text/event-stream, and sets stream: true in the body. The companion DecodeSse decodes one event — its members are associated functions, each handed what it needs: the config for response_specs, the head and the config for metadata, and event, head, config, and context for decode_event:

impl<T: StreamableChat> OverrideEndpoint for Stream<T> {
    type Endpt = T;
    type Output = StreamIterator<ChatResponseChunk>;
    type Error = GrokError;
    type Decoder = SseDecoder;
    type Requires = <T as Endpoint>::Requires;
    // ... endpoint_ref / endpoint, accept = "text/event-stream", body with `stream: true` ...
}

impl<T: StreamableChat> DecodeSse for Stream<T> {
    type Item = ChatResponseChunk;
    type Metadata = ();

    // The fixture plan: one model per event, `[DONE]` a protocol marker rather
    // than a payload, and the error envelope a pre-stream 4xx carries.
    fn response_specs(config: &Self::ApiConfig) -> Option<StreamSpecs> {
        let codec = config.json_lib().erase();
        Some(EventStreamSpec::sse::<ChatResponseChunk>(codec)
            .with_sentinels(&["[DONE]"])
            .with_error_envelope::<GrokErrorWire>(codec))
    }

    fn decode_event(
        event: ServerSentEvent<String>, _headers: &ResponseHead,
        config: &Self::ApiConfig, _context: ApiContext,
    ) -> Result<Option<Self::Item>, ResponseError<Self::Error>> {
        match event.data {
            Some(data) if data == "[DONE]" => Ok(None),   // terminal sentinel: consume it, the stream ends with the body
            Some(data) => Ok(Some(config.json_lib().decode_from_slice(data.as_bytes())?)),
            None => Ok(None),                              // a comment-only or metadata-only event
        }
    }

    fn decode_error(
        headers: ResponseHead, body: Vec<u8>, config: &Self::ApiConfig,
        _fallback: ResponseError<Self::Error>,
    ) -> ResponseError<Self::Error> {
        crate::decoders::build_api_error::<Self::Error>(&body, config, headers.status)
    }
}

Three defaulted members round out DecodeSse. type Metadata (with fn metadata(headers, config)) is the per-stream value the handle’s metadata() returns — header-derived information such as rate limits, () by default. validate_response admits a response by media type: text/event-stream in any case and with any parameters, so Text/Event-Stream; charset=utf-8 passes — override it for a service that mislabels its stream. And the spec builder has two more knobs beside with_sentinels: with_events(&[("name", Shape)]) declares a distinct payload model per event name, and without_error_envelope() closes the plan for a service whose pre-stream errors carry no model.

Consuming it is the same as any stream handle:

let mut stream = api.query(request.stream()).await?;
while let Some(chunk) = stream.try_next().await? {
    if let Some(text) = chunk.first_text() {
        print!("{text}");
    }
}

Streaming a buffered format: document streams

JSON is a buffered format — a codec decodes one complete document from a slice — and yet APIs stream it all the time: a large export served as one […] array, Docker’s progress objects one after another, NDJSON. Streaming such a body is a framing question, not a codec question: cut the body into complete documents, and hand each one to the codec you already hold. DocumentStreamDecoder<Framing> does the cutting; you decode.

The framing is a type. JsonArray consumes the [, the commas, and the ] and delivers the elements as they complete, so a multi-megabyte array streams element by element. JsonDocuments delivers whitespace-separated values — which includes NDJSON, with or without its final newline. Lines delivers one document per terminated line and skips blank keep-alives. JsonSeq frames RFC 7464 text sequences. The framer works from JSON’s lexical grammar alone (strings and their escapes, brackets, whitespace); it never validates or decodes a document, so any JSON codec fits, and none is named by the framework.

The decode step comes from one of two places, exactly as it does for BodyDecoder<S>. An endpoint can implement DecodeDocuments itself — the DecodeSse shape, with decode_document(frame: &[u8], headers, config, context) receiving one complete document. More often a client crate writes one strategy and every document stream in the crate becomes a declaration:

pub struct ViaJsonLib;

impl<Endpt, T> DecodeDocumentStrategy<Endpt> for ViaJsonLib
where
    Endpt: Endpoint<Output = StreamIterator<T>, ApiConfig = ParcelConfig, Error = ParcelError>,
    T: for<'de> WireDecode<'de> + WireShape + Unpin,
{
    type Item = T;
    type Metadata = ();

    fn response_specs(framing: StreamFraming, config: &ParcelConfig) -> Option<StreamSpecs> {
        let codec = config.json_lib().erase();
        Some(EventStreamSpec::framed::<T>(framing, codec).with_error_envelope::<ParcelErrorWire>(codec))
    }

    fn decode_error(headers: ResponseHead, body: Vec<u8>, config: &ParcelConfig,
                    _fallback: ResponseError<ParcelError>) -> ResponseError<ParcelError> {
        build_api_error::<ParcelError>(&body, config, headers.status)
    }

    fn decode_document(frame: &[u8], _headers: &ResponseHead, config: &ParcelConfig,
                       _context: ApiContext) -> Result<Option<T>, ResponseError<ParcelError>> {
        Ok(Some(config.json_lib().decode_from_slice(frame)?))
    }
}

impl Endpoint for ExportShipments {
    // ... ApiConfig ...
    type Output  = StreamIterator<Shipment>;
    // ... Error ...
    type Decoder = DocumentStreamDecoder<JsonArray, ViaJsonLib>;
    // ... Requires, method, url ...
}

The item type is whatever the endpoint’s Output names, so the one strategy serves a stream of shipments and a stream of invoices alike; the default strategy, ViaEndpoint, forwards to the endpoint’s own DecodeDocuments impl, the same slot BodyDecoder forwards to DecodeBody. response_specs receives the framing from the decoder — the fixture plan the test framework masks a recording under cannot name a framing the decoder does not cut. Admission rejects a non-2xx response and buffers its body for decode_error, so the pre-stream error path below applies here too; Content-Type is deliberately not checked, since a root array arrives as application/json and line-delimited JSON has several spellings.

Two limits are worth knowing. The framer finds boundaries and trusts them: JSON broken in a way that moves a boundary — an unescaped quote — misframes everything after it, and the stream ends at the codec’s first failure rather than resynchronising. And the frame cap bounds one document, which in the array framing is one element: an element larger than the cap fails where a buffered body of the same size would not.

Pre-stream errors: decode_error

This is the footgun from the decoder chapter. A stream decoder runs the SSE or document framing on the success path, but an error response (a 4xx before any events) isn’t framed data — so override decode_error to run the same shared build_api_error your body decoder uses. Skip it and the error body is silently lost. grok’s SSE, TTS, and download decoders all delegate to the one build_api_error, so a rate-limited streaming call surfaces the same typed error as a non-streaming one. The chunked decoder is the exception: DecodeChunked has no decode_error hook, and its default validate_response admits every status, so a chunked endpoint that can fail before its first chunk handles the status inside decode_chunk.

Binary downloads

A binary download sets Output = BinaryStream and Decoder = BytestreamDecoder, and implements DecodeBytestream. The happy path does zero buffering — bytes flow from the transport straight through the BinaryStream (which implements AsyncRead / Stream) to your consumer:

impl DecodeBytestream for PostTts {
    type Headers = ();   // or a TypedHeader struct to surface response headers
}

Set type Headers to a struct implementing TypedHeader to expose typed response headers (content type, length, custom x-* headers) alongside the byte stream; the decoder parses them off the response head and makes them available via stream.headers() and stream.content_type(). A BinaryStream derefs to the Bytestream underneath, so a content-coded download is decoded on the way in with .with_codec(..) and one of the capie_* decoders.

Resumable downloads

A download that must survive a mid-transfer disconnect sets Output = FileStream (or FileStream<Headers> for typed response headers), Decoder = ResumeDownloadDecoder, and implements ResumableDownload; the framework supplies everything else. grok’s pre-signed content URLs are served by object storage that honours Range, so its download is the shape:

impl Endpoint for DownloadContent {
    type Output = FileStream;
    type Decoder = ResumeDownloadDecoder;
    // ... GET on the absolute URL, Auth::NONE
}

impl ResumableDownload for DownloadContent {
    type Headers = ();

    fn decode_error(headers: &ResponseHead, body: &[u8], config: &Self::ApiConfig,
                    _fallback: ResponseError<Self::Error>) -> ResponseError<Self::Error> {
        crate::decoders::build_api_error::<Self::Error>(body, config, headers.status)
    }
}

On a break the driver re-sends the endpoint with Range from the absolute byte offset already delivered and If-Range carrying the validator it captured from the opening response, and it requires a 206 Partial Content continuing exactly there — a resource that changed mid-download, or a server that answers 200, ends the stream rather than splicing bytes. An opening request that itself carried a Range resumes relative to that base and completes at the end of what was requested. The trait’s defaulted members tune the policy: max_retries, max_reconnects, backoff(attempt, rng), idle_timeout, resume_conditional (which validator the resume guards with), and prepare_for_resume(&mut self, opening_headers) for an endpoint that must adjust itself before the second request. The FileStream it yields reads chunk by chunk (next_chunk, try_next_chunk, into_chunks() for a StreamIterator<Vec<u8>>), reports total_len() and suggested_filename() from the opening response’s FileStreamMeta, and on std sinks itself with write_to(writer), write_to_with_progress(..), or download_to_file(path). S3’s GetObjectResumable is the same decoder over an authenticated, signed endpoint.

The frame cap

The framing engine enforces a per-frame ceiling, so one malformed frame cannot grow without bound. The default is 32 MiB (32 << 20), read back with capi_base_decoders::max_frame_bytes(); it is process-global, set once at startup:

capi_base_decoders::set_max_frame_bytes(64 << 20);   // raise to 64 MiB; 0 = unlimited

A frame past the limit fails with capi_base_decoders::StreamError::FrameTooLarge { limit } (the full path matters: the prelude’s StreamError is capi_core’s, a different enum with only IncompleteEof). It’s global, not per-endpoint (and a no-op on targets without pointer-width atomics, fixed at 32 MiB), so set it deliberately and once.

The layers underneath

Three layers cooperate, and knowing the split helps when debugging: the transport delivers bytes as a Bytestream; a shared framing engine slices them into frames (respecting the cap); and your companion trait turns one frame into one Item. You only ever write the top layer — decode_event / decode_document / decode_chunk — and the engine handles buffering, the cap, and back-pressure.

The next chapters compose on this foundation: pagination is a stream of pages, and resumable streams wrap one in a reconnect policy.