{"record":{"id":"dbb95cf380ec7f77","repo":"risingwavelabs/risingwave","slug":"unconnected-worker-node","errorCode":null,"errorMessage":"unconnected worker node {}","messagePattern":"unconnected worker node (.+?)","errorType":"error_code","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/meta/src/barrier/rpc.rs","lineNumber":1298,"sourceCode":"        let mut node_need_collect = NodeToCollect::new();\n        let table_ids_to_sync = table_ids_to_sync.collect_vec();\n\n        node_actors.iter()\n            .try_for_each(|(worker_id, actor_ids_to_collect)| {\n                assert!(!actor_ids_to_collect.is_empty(), \"empty actor_ids_to_collect on worker {worker_id} in node_actors {node_actors:?}\");\n                let table_ids_to_sync = if nodes_to_sync_table.contains(worker_id) {\n                    table_ids_to_sync.clone()\n                } else {\n                    vec![]\n                };\n\n                let node = if let Some((_, worker_state)) = self.workers.get(worker_id)\n                    &&\n                    let WorkerNodeState::Connected { control_stream, .. } = worker_state\n                {\n                    control_stream\n                } else {\n                    return Err(anyhow!(\"unconnected worker node {}\", worker_id).into());\n                };\n\n                {\n                    let mutation = mutation.clone();\n                    let barrier = Barrier {\n                        epoch: Some(risingwave_pb::data::Epoch {\n                            curr: barrier_info.curr_epoch(),\n                            prev: barrier_info.prev_epoch(),\n                        }),\n                        mutation: mutation.clone().map(|_| BarrierMutation { mutation }),\n                        tracing_context: TracingContext::from_span(barrier_info.curr_epoch.span())\n                            .to_protobuf(),\n                        kind: barrier_info.kind.to_protobuf() as i32,\n                    };\n\n                    node.handle\n                        .request_sender\n                        .send(StreamingControlStreamRequest {","sourceCodeStart":1280,"sourceCodeEnd":1316,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/barrier/rpc.rs#L1280-L1316","documentation":"`inject_barrier` sends barrier mutations to each worker via its control stream, looked up in `self.workers`. A worker is only usable when its `WorkerNodeState::Connected` holds a control stream; otherwise this error is returned. It means the meta service tried to send a barrier to a worker that is not currently connected (never connected, disconnected, or draining).","triggerScenarios":"`inject_barrier` addressing worker_id whose entry in `self.workers` is absent, or whose state is not `WorkerNodeState::Connected` (e.g. disconnected during the barrier).","commonSituations":"Compute/foreground node crash or network partition mid-barrier; worker just registered but stream not yet established; cluster scale-in removing a node while barriers are in flight; stale worker ids referenced during recovery.","solutions":["Check the worker's liveness: it likely crashed or lost its gRPC stream; restart the worker node.","Verify network connectivity between meta node and the worker (host/port in worker host field).","Let barrier recovery run — the failed barrier will be retried once workers reconnect.","If a stale worker id is targeted, remove the stale worker registration from the cluster."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Check worker connectivity before injecting barriers\nasync fn worker_connected(workers: &HashMap<WorkerId, WorkerNodeState>, id: WorkerId) -> bool {\n    matches!(\n        workers.get(&id),\n        Some(WorkerNodeState::Connected { .. })\n    )\n}","typeGuard":"// Rust: narrow to the Connected variant\nfn control_stream_of(state: &WorkerNodeState) -> Option<&ControlStreamHandle> {\n    if let WorkerNodeState::Connected { control_stream, .. } = state {\n        Some(control_stream)\n    } else {\n        None\n    }\n}","tryCatchPattern":"match inject_barrier(mutation).await {\n    Err(e) if e.to_string().contains(\"unconnected worker node\") => {\n        // wait for reconnection, then rely on barrier retry\n        tokio::time::sleep(Duration::from_secs(1)).await;\n    }\n    other => other?,\n}","preventionTips":["Monitor worker liveness/heartbeats before scheduling barrier injections.","Filter target workers to Connected state at barrier build time.","Alert on frequent worker disconnects; they usually indicate node crashes or network faults."],"tags":["worker","connection","barrier","network"],"backgroundTag":"resource-not-found","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"}