{"record":{"id":"84dfabfa3b45f416","repo":"vectordotdev/vector","slug":"event-forward-rx-should-not-close-first","errorCode":null,"errorMessage":"event forward rx should not close first","messagePattern":"event forward rx should not close first","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/components/validation/runner/io.rs","lineNumber":57,"sourceCode":"#[tonic::async_trait]\nimpl VectorService for EventForwardService {\n    async fn push_events(\n        &self,\n        request: tonic::Request<PushEventsRequest>,\n    ) -> Result<tonic::Response<PushEventsResponse>, Status> {\n        let events = request\n            .into_inner()\n            .events\n            .into_iter()\n            .map(|wrapper| {\n                Event::try_from(wrapper).expect(\"validation events are encoded by Vector\")\n            })\n            .collect();\n\n        self.tx\n            .send(events)\n            .await\n            .expect(\"event forward rx should not close first\");\n\n        Ok(tonic::Response::new(PushEventsResponse {}))\n    }\n\n    async fn health_check(\n        &self,\n        _: tonic::Request<HealthCheckRequest>,\n    ) -> Result<tonic::Response<HealthCheckResponse>, Status> {\n        let message = HealthCheckResponse {\n            status: ServingStatus::Serving.into(),\n        };\n\n        Ok(tonic::Response::new(message))\n    }\n}\n\npub struct InputEdge {\n    #[allow(dead_code)]","sourceCodeStart":39,"sourceCodeEnd":75,"githubUrl":"https://github.com/vectordotdev/vector/blob/bdb87aeaa4c4ff27c0ba643c1c77b21bf2ef4013/src/components/validation/runner/io.rs#L39-L75","documentation":"push_events sends the collected events over an mpsc channel to the validation runner. The send is expected to succeed because the receiving end (event forward rx) is held by the runner for the server's lifetime; a panic means the receiver was dropped first, i.e. the runner shut down while the gRPC server was still pushing events.","triggerScenarios":"tx.send(events).await returns Err (RecvError) because the runner dropped the forwarding receiver — runner shutdown racing ahead of the gRPC output server, or premature drop of RunnerOutput plumbing.","commonSituations":"Shutdown-ordering bugs in the validation runner; a panic in the runner task that owned the rx, causing the server to panic on the next push.","solutions":["Check the runner-side task logs for an earlier panic that dropped the receiver.","Ensure shutdown coordination (input_task_coordinator/output_task_coordinator) closes the gRPC server before dropping rx.","Handle the send error gracefully (return a gRPC Status::unavailable) rather than panicking."],"exampleFix":"// before\nself.tx.send(events).await.expect(\"event forward rx should not close first\");\n// after\nself.tx.send(events).await.map_err(|_| tonic::Status::unavailable(\"runner shut down\"))?;","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"if self.tx.send(events).await.is_err() {\n    return Ok(()) /* or Status::unavailable */; // runner already shut down\n}","preventionTips":["Enforce shutdown ordering: close the gRPC server before dropping the forwarding receiver.","Check runner task logs for earlier panics that would drop the receiver.","Treat channel send failures as graceful shutdown signals rather than panics."],"tags":["grpc","channel","shutdown","panic"],"backgroundTag":"broken-pipe","analyzedSha":"bdb87aeaa4c4ff27c0ba643c1c77b21bf2ef4013","analyzedAt":"2026-09-16T02:53:35.741Z","contentChangedAt":"2026-09-16T02:53:35.741Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}