risingwavelabs/risingwave · error

failed to collect barrier: notifier dropped

Error message

failed to collect barrier: notifier dropped

What it means

`wait_collection` waits on barrier-collection receivers; each receiver's counterpart (the `CollectionHandle`/notifier) sends the result. If the sender side is dropped before completing (Rx closed with Err), the wait can never succeed, so it is reported as "notifier dropped". The library throws it because a dropped notifier means the barrier collection was aborted (typically due to a failure/abort path) rather than completed.

Solutions

1. Treat this as an expected failure-path signal: catch it and re-drive recovery (the barrier manager will retry/abort the barrier).
2. Ensure every spawned `CollectionHandle` is completed or explicitly failed before being dropped, never silently discarded.
3. If it appears spuriously, check for early-return/panic paths in the barrier injection task that drop handles while collectors still wait.
4. Inspect meta logs for the underlying barrier failure that caused the notifier to be dropped.

Example fix

// before
let result = wait_collection(receivers).await?;
// after
match wait_collection(receivers).await {
    Ok(()) => {},
    Err(e) if e.to_string().contains("notifier dropped") => {
        // barrier was aborted; rely on recovery instead of surfacing to user
        tracing::warn!("barrier collection aborted: {e:#}");
    }
    Err(e) => return Err(e),
}

// Rust: treat the anyhow error text as the discriminator
fn is_notifier_dropped(err: &MetaError) -> bool {
    err.to_string().contains("notifier dropped")
}

match wait_collection(receivers).await {
    Ok(()) => {}
    Err(e) if e.to_string().contains("notifier dropped") => {
        tracing::warn!("barrier aborted, relying on recovery: {e:#}");
    }
    Err(e) => return Err(e),
}

When it happens

Trigger: Calling `wait_collection(receivers)` where one or more `CollectionReceiver`s' sending halves were dropped — e.g. after `notify_failed`/abort drops collection handles, or a handle is discarded without ever calling complete.

Common situations: Barrier manager failure recovery: a barrier collection fails and handles are dropped while another task is still awaiting them; tests like `test_waits_for_all_collection_notifiers_after_failure` exercise exactly this; leaks where a notifier is dropped without resolving.

Related errors

• anyhow!(message.to_owned())
• barrier reader closed unexpectedly
• batch refresh job has no snapshot backfill info
• cannot create batch refresh job while database barrier is…
• database unavailable

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

Appendix: source

Thrown at src/meta/src/barrier/notifier.rs:29

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use anyhow::anyhow;
use futures::future::join_all;
use tokio::sync::oneshot;

use crate::{MetaError, MetaResult};

pub(crate) type CollectionReceiver = oneshot::Receiver<MetaResult<()>>;
pub(crate) type StartReceiver = oneshot::Receiver<MetaResult<Vec<CollectionReceiver>>>;

pub(crate) async fn wait_collection(receivers: Vec<CollectionReceiver>) -> MetaResult<()> {
    let mut first_error = None;
    for result in join_all(receivers).await {
        let result = match result {
            Ok(result) => result,
            Err(_) => Err(anyhow!("failed to collect barrier: notifier dropped").into()),
        };
        if let Err(err) = result
            && first_error.is_none()
        {
            first_error = Some(err);
        }
    }
    match first_error {
        Some(err) => Err(err),
        None => Ok(()),
    }
}

/// Used for notifying the status of a scheduled command/barrier.
#[derive(Debug)]
pub(crate) struct Notifier {
    started: oneshot::Sender<MetaResult<Vec<CollectionReceiver>>>,
}

View on GitHub (pinned to 6469eb736d)