What causes "end of response stream" in risingwavelabs/risingwave?

Re-establish the stream via BidiStreamHandle::initialize and resend pending requests.

risingwavelabs/risingwave · error · RpcError

end of response stream

Error message

end of response stream

What it means

BidiStreamReceiver::next_response found the response stream exhausted: the boxed stream of Result<RSP> yielded None, meaning the server closed the stream (possibly after an error the stream itself already delivered). The None is converted to an anyhow error 'end of response stream'.

Source

Thrown at src/rpc_client/src/lib.rs:241

impl<REQ> BidiStreamSender<REQ> {
    pub async fn send_request<R: Into<REQ>>(&mut self, request: R) -> Result<()> {
        self.tx
            .send(request.into())
            .await
            .map_err(|_| anyhow!("unable to send request {}", type_name::<REQ>()).into())
    }
}

pub struct BidiStreamReceiver<RSP> {
    pub stream: Peekable<BoxStream<'static, Result<RSP>>>,
}

impl<RSP> BidiStreamReceiver<RSP> {
    pub async fn next_response(&mut self) -> Result<RSP> {
        self.stream
            .next()
            .await
            .ok_or_else(|| anyhow!("end of response stream"))?
    }
}

pub struct BidiStreamHandle<REQ, RSP> {
    pub request_sender: BidiStreamSender<REQ>,
    pub response_stream: BidiStreamReceiver<RSP>,
}

impl<REQ, RSP> Debug for BidiStreamHandle<REQ, RSP> {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.write_str(type_name::<Self>())
    }
}

impl<REQ, RSP> BidiStreamHandle<REQ, RSP> {
    pub fn for_test(
        request_sender: Sender<REQ>,
        response_stream: BoxStream<'static, Result<RSP>>,

View on GitHub (pinned to 6469eb736d)

Solutions

Re-establish the stream via BidiStreamHandle::initialize and resend pending requests.
Check server logs for stream termination cause (panic, cancellation, shutdown).
Add retry/backoff around the RPC session in callers that expect long-lived streams.

Example fix

// before
self.stream.next().await.ok_or_else(|| anyhow!("end of response stream"))?
// after (caller)
match handle.response_stream.next_response().await {
    Ok(rsp) => rsp,
    Err(_) => { let handle = BidiStreamHandle::initialize(&client, first_request).await?; /* retry */ }
}

Defensive patterns

Strategy: retry

Try / catch

match handle.response_stream.next_response().await {
    Ok(rsp) => rsp,
    Err(e) if e.to_string().contains("end of response stream") => {
        handle = BidiStreamHandle::initialize(&client, first_req).await?; // retry
    }
    Err(e) => return Err(e.into()),
}

Prevention

Wrap long-lived bidi sessions with automatic reconnection and backoff.
Watch server logs for stream cancellation to distinguish crash vs. graceful end.
Set gRPC keepalive so half-open streams terminate predictably.

When it happens

Trigger: Calling next_response on a BidiStreamReceiver after the server finished the RPC stream; also occurs inside initialize when first_response is polled on an immediately-closing stream.

Common situations: Server-side task panic or graceful completion before answering all requests; network drop causing tonic to end the stream; client polling a stream it already saw terminate.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/af1ecc4075aa21bb. Report an issue: GitHub.