{"record":{"id":"a39f9fc77f007e28","repo":"risingwavelabs/risingwave","slug":"unable-to-send-first-request-of","errorCode":null,"errorMessage":"unable to send first request of {}","messagePattern":"unable to send first request of (.+?)","errorType":"exception","errorClass":"RpcError","httpStatus":null,"severity":"error","filePath":"src/rpc_client/src/lib.rs","lineNumber":284,"sourceCode":"        }\n    }\n\n    pub async fn initialize<\n        F: FnOnce(Receiver<REQ>) -> Fut,\n        St: Stream<Item = Result<RSP>> + Send + Unpin + 'static,\n        Fut: Future<Output = Result<St>> + Send,\n        R: Into<REQ>,\n    >(\n        first_request: R,\n        init_stream_fn: F,\n    ) -> Result<(Self, RSP)> {\n        let (request_sender, request_receiver) = channel(DEFAULT_BUFFER_SIZE);\n\n        // Send initial request in case of the blocking receive call from creating streaming request\n        request_sender\n            .send(first_request.into())\n            .await\n            .map_err(|_err| anyhow!(\"unable to send first request of {}\", type_name::<REQ>()))?;\n\n        let mut response_stream = init_stream_fn(request_receiver).await?;\n\n        let first_response = response_stream\n            .next()\n            .await\n            .ok_or_else(|| anyhow!(\"get empty response from first request\"))??;\n\n        Ok((\n            Self {\n                request_sender: BidiStreamSender { tx: request_sender },\n                response_stream: BidiStreamReceiver {\n                    stream: response_stream.boxed().peekable(),\n                },\n            },\n            first_response,\n        ))\n    }","sourceCodeStart":266,"sourceCodeEnd":302,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/rpc_client/src/lib.rs#L266-L302","documentation":"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.","triggerScenarios":"Calling BidiStreamHandle::initialize when init_stream_fn's underlying gRPC streaming call fails immediately (server unreachable, endpoint rejected), dropping the request receiver.","commonSituations":"Connector/meta service down or restarting; wrong service address/port configuration; auth or version rejection on the streaming method.","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."],"exampleFix":"// before\nrequest_sender.send(first_request.into()).await.map_err(|_err| anyhow!(\"unable to send first request of {}\", type_name::<REQ>()))?;\n// after (caller-level)\nlet handle = loop {\n    match BidiStreamHandle::initialize(&client, first.clone()).await {\n        Ok(h) => break h,\n        Err(e) => { warn!(\"init failed: {e}\"); sleep(Duration::from_secs(1)).await; }\n    }\n};","handlingStrategy":"retry","validationCode":"// before initialize\nassert!(service_reachable(connector_endpoint).await, \"connector service must be reachable\");","typeGuard":null,"tryCatchPattern":"let handle = retry(BoundedBackoff::from_secs(1), || async {\n    BidiStreamHandle::initialize(&client, first.clone()).await\n}).await?;","preventionTips":["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."],"tags":["rpc","streaming","grpc","channel-closed","startup"],"backgroundTag":"connection-refused","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"}