risingwavelabs/risingwave · critical

notification stopped or uninitialized

Error message

notification stopped or uninitialized

What it means

MetadataManager::changed awaits the next notification from its receiver with .expect(...). The panic fires when the receiver returns None — all notification senders were dropped (notification service shut down) or the channel was never initialized. This is a fatal meta-side invariant: the manager cannot observe worker changes without the notification stream.

Solutions

1. If seen during meta shutdown, ignore it — the panic accompanies process exit.
2. Fix startup ordering: create the notification service and hold its sender before MetadataManager spawns changed()-driven tasks.
3. If the meta node keeps running, capture the panic backtrace and check for early drops of the notification sender; fix ownership so the sender outlives the manager.
4. Replace the expect with a graceful error/log if the manager should tolerate service shutdown, and report persistent occurrences as a bug.

Example fix

// before
let notification = self.rx.recv().await.expect("notification stopped or uninitialized");
// after
let notification = self.rx.recv().await.ok_or_else(||
    anyhow!("metadata notification stream closed; notification service stopped or uninitialized")
)?;

// ensure the notification channel is live before entering the manager loop
assert!(!rx.is_closed(), "notification channel closed before MetadataManager start");

// supervisor around the manager loop
while let Ok(()) = shutdown_rx.changed().await {
    match metadata_manager.changed().await {
        Ok(change) => handle(change).await,
        Err(_) => {
            // notification service stopped; exit loop gracefully instead of panicking
            break;
        }
    }
}

When it happens

Trigger: changed() polls while the meta notification service is shutting down, after the sender half is dropped, or when MetadataManager was constructed without wiring the notification channel (uninitialized rx).

Common situations: Meta node shutdown/failover while background tasks still call changed(); a startup ordering bug where the manager starts before the notification service; test harnesses constructing MetadataManager without a notification sender.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• internal error: entered unreachable code
• failed to parse relation definition
• Failed to retrieve fragment description: fragment
• invalid parallelism
• job fragments should exist for streaming job

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/cbae13c48c588605. Report an issue: GitHub.

Appendix: source

Thrown at src/meta/src/manager/metadata.rs:123

                    is_streaming.then_some((node.id, node))
                })
                .collect(),
            rx,
            meta_manager: Some(meta_manager),
        })
    }

    pub(crate) fn current(&self) -> &HashMap<WorkerId, WorkerNode> {
        &self.worker_nodes
    }

    pub(crate) async fn changed(&mut self) -> ActiveStreamingWorkerChange {
        loop {
            let notification = self
                .rx
                .recv()
                .await
                .expect("notification stopped or uninitialized");
            fn is_target_worker_node(worker: &WorkerNode) -> bool {
                worker.r#type == WorkerType::ComputeNode as i32
                    && worker.property.as_ref().unwrap().is_streaming
            }
            match notification {
                LocalNotification::WorkerNodeDeleted(worker) => {
                    let is_target_worker_node = is_target_worker_node(&worker);
                    let Some(prev_worker) = self.worker_nodes.remove(&worker.id) else {
                        if is_target_worker_node {
                            warn!(
                                ?worker,
                                "notify to delete an non-existing streaming compute worker"
                            );
                        }
                        continue;
                    };
                    if !is_target_worker_node {
                        warn!(

View on GitHub (pinned to 6469eb736d)