{"record":{"id":"af1ecc4075aa21bb","repo":"risingwavelabs/risingwave","slug":"end-of-response-stream-af1ecc","errorCode":null,"errorMessage":"end of response stream","messagePattern":"end of response stream","errorType":"exception","errorClass":"RpcError","httpStatus":null,"severity":"error","filePath":"src/rpc_client/src/lib.rs","lineNumber":241,"sourceCode":"impl<REQ> BidiStreamSender<REQ> {\n    pub async fn send_request<R: Into<REQ>>(&mut self, request: R) -> Result<()> {\n        self.tx\n            .send(request.into())\n            .await\n            .map_err(|_| anyhow!(\"unable to send request {}\", type_name::<REQ>()).into())\n    }\n}\n\npub struct BidiStreamReceiver<RSP> {\n    pub stream: Peekable<BoxStream<'static, Result<RSP>>>,\n}\n\nimpl<RSP> BidiStreamReceiver<RSP> {\n    pub async fn next_response(&mut self) -> Result<RSP> {\n        self.stream\n            .next()\n            .await\n            .ok_or_else(|| anyhow!(\"end of response stream\"))?\n    }\n}\n\npub struct BidiStreamHandle<REQ, RSP> {\n    pub request_sender: BidiStreamSender<REQ>,\n    pub response_stream: BidiStreamReceiver<RSP>,\n}\n\nimpl<REQ, RSP> Debug for BidiStreamHandle<REQ, RSP> {\n    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {\n        f.write_str(type_name::<Self>())\n    }\n}\n\nimpl<REQ, RSP> BidiStreamHandle<REQ, RSP> {\n    pub fn for_test(\n        request_sender: Sender<REQ>,\n        response_stream: BoxStream<'static, Result<RSP>>,","sourceCodeStart":223,"sourceCodeEnd":259,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/rpc_client/src/lib.rs#L223-L259","documentation":"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'.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"// before\nself.stream.next().await.ok_or_else(|| anyhow!(\"end of response stream\"))?\n// after (caller)\nmatch handle.response_stream.next_response().await {\n    Ok(rsp) => rsp,\n    Err(_) => { let handle = BidiStreamHandle::initialize(&client, first_request).await?; /* retry */ }\n}","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"match handle.response_stream.next_response().await {\n    Ok(rsp) => rsp,\n    Err(e) if e.to_string().contains(\"end of response stream\") => {\n        handle = BidiStreamHandle::initialize(&client, first_req).await?; // retry\n    }\n    Err(e) => return Err(e.into()),\n}","preventionTips":["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."],"tags":["rpc","streaming","grpc","stream-closed"],"backgroundTag":"empty-response-body","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}