{"record":{"id":"746e03e34771b097","repo":"risingwavelabs/risingwave","slug":"failed-to-collect-barrier-notifier-dropped","errorCode":null,"errorMessage":"failed to collect barrier: notifier dropped","messagePattern":"failed to collect barrier: notifier dropped","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/barrier/notifier.rs","lineNumber":29,"sourceCode":"// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n// See the License for the specific language governing permissions and\n// limitations under the License.\n\nuse anyhow::anyhow;\nuse futures::future::join_all;\nuse tokio::sync::oneshot;\n\nuse crate::{MetaError, MetaResult};\n\npub(crate) type CollectionReceiver = oneshot::Receiver<MetaResult<()>>;\npub(crate) type StartReceiver = oneshot::Receiver<MetaResult<Vec<CollectionReceiver>>>;\n\npub(crate) async fn wait_collection(receivers: Vec<CollectionReceiver>) -> MetaResult<()> {\n    let mut first_error = None;\n    for result in join_all(receivers).await {\n        let result = match result {\n            Ok(result) => result,\n            Err(_) => Err(anyhow!(\"failed to collect barrier: notifier dropped\").into()),\n        };\n        if let Err(err) = result\n            && first_error.is_none()\n        {\n            first_error = Some(err);\n        }\n    }\n    match first_error {\n        Some(err) => Err(err),\n        None => Ok(()),\n    }\n}\n\n/// Used for notifying the status of a scheduled command/barrier.\n#[derive(Debug)]\npub(crate) struct Notifier {\n    started: oneshot::Sender<MetaResult<Vec<CollectionReceiver>>>,\n}","sourceCodeStart":11,"sourceCodeEnd":47,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/barrier/notifier.rs#L11-L47","documentation":"`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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Treat this as an expected failure-path signal: catch it and re-drive recovery (the barrier manager will retry/abort the barrier).","Ensure every spawned `CollectionHandle` is completed or explicitly failed before being dropped, never silently discarded.","If it appears spuriously, check for early-return/panic paths in the barrier injection task that drop handles while collectors still wait.","Inspect meta logs for the underlying barrier failure that caused the notifier to be dropped."],"exampleFix":"// before\nlet result = wait_collection(receivers).await?;\n// after\nmatch wait_collection(receivers).await {\n    Ok(()) => {},\n    Err(e) if e.to_string().contains(\"notifier dropped\") => {\n        // barrier was aborted; rely on recovery instead of surfacing to user\n        tracing::warn!(\"barrier collection aborted: {e:#}\");\n    }\n    Err(e) => return Err(e),\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":"// Rust: treat the anyhow error text as the discriminator\nfn is_notifier_dropped(err: &MetaError) -> bool {\n    err.to_string().contains(\"notifier dropped\")\n}","tryCatchPattern":"match wait_collection(receivers).await {\n    Ok(()) => {}\n    Err(e) if e.to_string().contains(\"notifier dropped\") => {\n        tracing::warn!(\"barrier aborted, relying on recovery: {e:#}\");\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Always complete or explicitly fail every CollectionHandle before dropping it.","In failure paths, notify all collectors instead of letting handles go out of scope.","Add tests mirroring test_waits_for_all_collection_notifiers_after_failure for any new abort path."],"tags":["barrier","channel-closed","async","meta"],"backgroundTag":"broken-pipe","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}