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'.

Solutions

1. Re-establish the stream via BidiStreamHandle::initialize and resend pending requests.
2. Check server logs for stream termination cause (panic, cancellation, shutdown).
3. 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 */ }
}

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()),
}

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

• get different response epoch to commit epoch
• get empty response from first request
• should get Commit response but get
• should get Sync response but get
• unable to send first request of

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

Appendix: 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)