{"record":{"id":"33f61224e5d3a511","repo":"risingwavelabs/risingwave","slug":"end-of-new-output-request","errorCode":null,"errorMessage":"end of new output request","messagePattern":"end of new output request","errorType":"error_code","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/dispatch.rs","lineNumber":146,"sourceCode":"    ) -> StreamResult<Vec<Output>> {\n        fn resolve_output(downstream_actor: ActorId, request: NewOutputRequest) -> Output {\n            let tx = match request {\n                NewOutputRequest::Local(tx) | NewOutputRequest::Remote(tx) => tx,\n            };\n            Output::new(downstream_actor, tx)\n        }\n        let mut outputs = Vec::with_capacity(downstream_actors.len());\n        for &downstream_actor in downstream_actors {\n            let output =\n                if let Some(request) = self.pending_new_output_requests.remove(&downstream_actor) {\n                    resolve_output(downstream_actor, request)\n                } else {\n                    loop {\n                        let (requested_actor, request) = self\n                            .new_output_request_rx\n                            .recv()\n                            .await\n                            .ok_or_else(|| anyhow!(\"end of new output request\"))?;\n                        if requested_actor == downstream_actor {\n                            break resolve_output(requested_actor, request);\n                        } else {\n                            assert!(\n                                self.pending_new_output_requests\n                                    .insert(requested_actor, request)\n                                    .is_none(),\n                                \"duplicated inflight new output requests from actor {}\",\n                                requested_actor\n                            );\n                        }\n                    }\n                };\n            outputs.push(output);\n        }\n        Ok(outputs)\n    }\n","sourceCodeStart":128,"sourceCodeEnd":164,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/dispatch.rs#L128-L164","documentation":"This anyhow error is raised in the DispatchExecutor's output collection when `new_output_request_rx.recv()` returns None, meaning all senders of the `new_output_request` mpsc channel have been dropped. The executor expected a new-output request from a downstream actor but the channel's sender side is gone, so no request can ever arrive. It indicates the actor/dispatcher lifecycle ended or was torn down unexpectedly while a downstream output resolution was still pending.","triggerScenarios":"Calling `collect_outputs` (via `add_dispatchers` or `pre_update_dispatcher`) while the `new_output_request_tx` senders have all been dropped, so `recv()` returns None instead of a `(requested_actor, request)` message.","commonSituations":"Upstream actor terminated or failed before sending the output request; actor rescale/dispatcher update racing with actor shutdown; misconfigured actor graph where the expected sender was never created; stream job cancellation during dispatcher rebalance.","solutions":["Check upstream actor logs for early termination or failure that dropped the new_output_request senders.","Verify the actor graph/dispatcher wiring so every downstream that calls collect_outputs has a live upstream sender.","Ensure actor shutdown ordering tears down the dispatcher executor before dropping senders, so collect_outputs is not invoked on a dead channel.","Retry the rescale/job operation after the failed dispatcher update; this is a lifecycle race rather than a data error."],"exampleFix":"// before: blindly resolving output even if channel is dead\nlet (requested_actor, request) = self.new_output_request_rx.recv().await.unwrap();\n// after: propagate a descriptive error and handle channel closure\nlet (requested_actor, request) = self\n    .new_output_request_rx\n    .recv()\n    .await\n    .ok_or_else(|| anyhow!(\"end of new output request\"))?;","handlingStrategy":"try-catch","validationCode":"// check channel liveness before collecting outputs\nif self.new_output_request_tx.is_closed() {\n    return Err(anyhow!(\"new_output_request channel closed before collect_outputs\"));\n}","typeGuard":null,"tryCatchPattern":"match self.new_output_request_rx.recv().await {\n    Some((actor, req)) => resolve_output(actor, req),\n    None => {\n        tracing::warn!(\"output request channel ended; aborting dispatcher update\");\n        return Err(anyhow!(\"end of new output request\"));\n    }\n}","preventionTips":["Keep a strong sender handle alive for the lifetime of collect_outputs.","Enforce shutdown ordering: finish dispatcher updates before dropping upstream actors.","Log actor termination events to correlate channel closure with collect_outputs calls.","Treat channel closure as a lifecycle error and propagate it instead of unwrap/expect."],"tags":["streaming","channel-closed","actor-lifecycle","rust"],"backgroundTag":"broken-pipe","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}