What causes "unable to send first request of {}" in risingwavelabs/risingwave?

Verify the target service address and that the connector/meta service is running, then retry initialize.

risingwavelabs/risingwave · error · RpcError

unable to send first request of {}

Error message

unable to send first request of {}

What it means

BidiStreamHandle::initialize failed to queue the first request: the initial request_sender.send() failed because the receiver passed to init_stream_fn dropped before accepting it, typically because establishing the streaming RPC itself failed.

Source

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

        }
    }

    pub async fn initialize<
        F: FnOnce(Receiver<REQ>) -> Fut,
        St: Stream<Item = Result<RSP>> + Send + Unpin + 'static,
        Fut: Future<Output = Result<St>> + Send,
        R: Into<REQ>,
    >(
        first_request: R,
        init_stream_fn: F,
    ) -> Result<(Self, RSP)> {
        let (request_sender, request_receiver) = channel(DEFAULT_BUFFER_SIZE);

        // Send initial request in case of the blocking receive call from creating streaming request
        request_sender
            .send(first_request.into())
            .await
            .map_err(|_err| anyhow!("unable to send first request of {}", type_name::<REQ>()))?;

        let mut response_stream = init_stream_fn(request_receiver).await?;

        let first_response = response_stream
            .next()
            .await
            .ok_or_else(|| anyhow!("get empty response from first request"))??;

        Ok((
            Self {
                request_sender: BidiStreamSender { tx: request_sender },
                response_stream: BidiStreamReceiver {
                    stream: response_stream.boxed().peekable(),
                },
            },
            first_response,
        ))
    }

View on GitHub (pinned to 6469eb736d)

Solutions

Verify the target service address and that the connector/meta service is running, then retry initialize.
Inspect init_stream_fn's own error (returned as the inner error of subsequent failures) for the gRPC failure cause.
Add a retry with backoff around stream initialization for transient unavailability.

Example fix

// before
request_sender.send(first_request.into()).await.map_err(|_err| anyhow!("unable to send first request of {}", type_name::<REQ>()))?;
// after (caller-level)
let handle = loop {
    match BidiStreamHandle::initialize(&client, first.clone()).await {
        Ok(h) => break h,
        Err(e) => { warn!("init failed: {e}"); sleep(Duration::from_secs(1)).await; }
    }
};

Defensive patterns

Strategy: retry

Validate before calling

// before initialize
assert!(service_reachable(connector_endpoint).await, "connector service must be reachable");

Try / catch

let handle = retry(BoundedBackoff::from_secs(1), || async {
    BidiStreamHandle::initialize(&client, first.clone()).await
}).await?;

Prevention

Health-check the target service before the first RPC of the process.
Configure retries with exponential backoff for stream initialization.
Validate service address/port config at startup with a probe connection.

When it happens

Trigger: Calling BidiStreamHandle::initialize when init_stream_fn's underlying gRPC streaming call fails immediately (server unreachable, endpoint rejected), dropping the request receiver.

Common situations: Connector/meta service down or restarting; wrong service address/port configuration; auth or version rejection on the streaming method.

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

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