{"record":{"id":"f5261fe2d47c5bbe","repo":"risingwavelabs/risingwave","slug":"none-msg-in-request","errorCode":null,"errorMessage":"None msg in request","messagePattern":"None msg in request","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/handle.rs","lineNumber":121,"sourceCode":"    }\n\n    pub(super) fn stop(&mut self) -> anyhow::Result<()> {\n        self.response_tx\n            .send(Ok(CoordinateResponse {\n                msg: Some(coordinate_response::Msg::Stopped(true)),\n            }))\n            .map_err(|_| anyhow!(\"failed to send the stopped response\"))\n    }\n\n    pub(super) fn poll_next_request(\n        &mut self,\n        cx: &mut Context<'_>,\n    ) -> Poll<anyhow::Result<coordinate_request::Msg>> {\n        let result = try {\n            let request = ready!(self.request_stream.try_poll_next_unpin(cx))\n                .ok_or_else(|| anyhow!(\"end of request stream\"))?\n                .map_err(anyhow::Error::from)?;\n            let request = request.msg.ok_or_else(|| anyhow!(\"None msg in request\"))?;\n            match &request {\n                coordinate_request::Msg::StartRequest(_)\n                | coordinate_request::Msg::Stop(_)\n                | coordinate_request::Msg::AlignInitialEpochRequest(_) => {}\n                coordinate_request::Msg::CommitRequest(request) => {\n                    if let Some(prev_epoch) = self.prev_epoch\n                        && request.epoch < prev_epoch\n                    {\n                        return Poll::Ready(Err(anyhow!(\n                            \"invalid commit epoch {}, prev_epoch {}\",\n                            request.epoch,\n                            prev_epoch\n                        )));\n                    }\n                    if request.metadata.is_none() {\n                        return Poll::Ready(Err(anyhow!(\"empty commit metadata\")));\n                    };\n                    self.prev_epoch = Some(request.epoch);","sourceCodeStart":103,"sourceCodeEnd":139,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/handle.rs#L103-L139","documentation":"The coordinator received a `CoordinateRequest` whose oneof `msg` field was unset (None). Protobuf oneof fields can be absent when a peer sends an empty/default message, and the coordinator cannot dispatch a message with no variant. This indicates a protocol violation by the sender.","triggerScenarios":"A `CoordinateRequest` arrives over the request stream with all optional `msg` variants unset — e.g. the client constructed the message without setting `msg`, or a version mismatch drops the field during serialization.","commonSituations":"Mixed RisingWave versions (older executor sends a message shape the current meta does not expect); a client bug building `CoordinateRequest { msg: None }`; corrupted or truncated gRPC frames decoded as empty messages.","solutions":["Check that all sink executors and meta nodes run the same RisingWave version (protobuf definitions in sync).","Inspect the sender code path that builds `CoordinateRequest` and ensure it always sets `msg`.","Reproduce with debug logging on the coordinate stream to capture the raw message before dispatch.","If a protobuf schema change is in flight, regenerate `prost` types consistently on both sides."],"exampleFix":"// before\nCoordinateRequest { msg: None }\n// after\nCoordinateRequest { msg: Some(coordinate_request::Msg::StartRequest(Default::default())) }","handlingStrategy":"validation","validationCode":"// Sender side: always set msg before sending\nlet request = CoordinateRequest { msg: Some(msg) };\nassert!(request.msg.is_some(), \"CoordinateRequest must carry a msg\");","typeGuard":null,"tryCatchPattern":"match handle.poll_next_request(cx) {\n    Poll::Ready(Err(e)) if e.to_string().contains(\"None msg in request\") => {\n        tracing::error!(\"peer sent empty CoordinateRequest; check version skew: {e:#}\");\n    }\n    other => { /* normal handling */ }\n}","preventionTips":["Always construct protobuf oneof messages with Some(variant)","Keep protobuf definitions in sync across all components","Add unit tests asserting every outgoing CoordinateRequest has a msg"],"tags":["rust","meta","protobuf","protocol-violation","sink-coordinator"],"backgroundTag":"protobuf-unmarshal-failed","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"}