risingwavelabs/risingwave · error · RpcError

unable to send request {}

Error message

unable to send request {}

What it means

BidiStreamSender::send_request failed because the mpsc channel's receiver half was dropped, so the request (named by its Rust type via type_name) could not be delivered over the bidirectional stream. The send error is swallowed and replaced by this anyhow message wrapped in RpcError.

Source

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

                    }
                }
            }
        )*
    }
}

pub const DEFAULT_BUFFER_SIZE: usize = 16;

pub struct BidiStreamSender<REQ> {
    tx: Sender<REQ>,
}

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>,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Treat this as a broken stream: call BidiStreamHandle::initialize again to establish a fresh stream and resend the request.
2. Check server (meta/connector) logs for the task that terminated the stream.
3. Stop sending on the old handle once any response-stream error is observed; guard with a closed flag in callers that loop.

Example fix

// before
self.tx.send(request.into()).await.map_err(|_| anyhow!("unable to send request {}", type_name::<REQ>()).into())
// after: propagate cause and advise re-initialization
self.tx.send(request.into()).await.map_err(|_| {
    anyhow!("unable to send request {} (stream closed; re-initialize the stream)", type_name::<REQ>())
})?.into()

match handle.request_sender.send_request(req).await {
    Err(e) if e.to_string().contains("unable to send request") => {
        handle = BidiStreamHandle::initialize(&client, first_req).await?;
        handle.request_sender.send_request(req).await?;
    }
    other => other?,
}

When it happens

Trigger: Calling send_request after the server-side stream task holding the receiver has finished or errored; also when awaiting send on a closed bounded (DEFAULT_BUFFER_SIZE) channel whose consumer stopped.

Common situations: Remote meta/connector node restarted mid-session; long-lived BidiStreamHandle reused after an RPC error on the response side; client keeps sending after seeing a stream error elsewhere.

Related errors

• unable to send first request of {}
• should get Sync response but get {:?}
• get different response epoch to commit epoch: {} {}
• should get Commit response but get {:?}
• end of response stream

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