{"record":{"id":"62e7901ae86b35bb","repo":"vectordotdev/vector","slug":"finalizer-must-have-been-set-up-for-acknowledgemen","errorCode":null,"errorMessage":"Finalizer must have been set up for acknowledgements","messagePattern":"Finalizer must have been set up for acknowledgements","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/gcp_pubsub.rs","lineNumber":642,"sourceCode":"            busy_flag.store(true, Ordering::Relaxed);\n        }\n        self.bytes_received.emit(ByteSize(response.size_of()));\n\n        let (batch, notifier) = BatchNotifier::maybe_new_with_receiver(self.acknowledgements);\n        let (events, ids) = self.parse_messages(response.received_messages, batch).await;\n\n        let count = events.len();\n        match self.out.send_batch(events).await {\n            Err(_) => emit!(StreamClosedError { count }),\n            Ok(()) => match notifier {\n                None => ack_ids\n                    .send(ids)\n                    .await\n                    .unwrap_or_else(|_| unreachable!(\"request stream never closes\")),\n                Some(notifier) => {\n                    finalizer\n                        .as_ref()\n                        .expect(\"Finalizer must have been set up for acknowledgements\")\n                        .add(ids, notifier);\n                    *pending_acks += 1;\n                }\n            },\n        }\n    }\n\n    async fn parse_messages(\n        &self,\n        response: Vec<proto::ReceivedMessage>,\n        batch: Option<BatchNotifier>,\n    ) -> (Vec<Event>, Vec<String>) {\n        let mut ack_ids = Vec::with_capacity(response.len());\n        let events = response\n            .into_iter()\n            .flat_map(|received| {\n                ack_ids.push(received.ack_id);\n                received","sourceCodeStart":624,"sourceCodeEnd":660,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/gcp_pubsub.rs#L624-L660","documentation":"In gcp_pubsub's streaming pull, each response batch sets up acknowledgement tracking with (batch, notifier) = BatchNotifier::maybe_new_with_receiver(self.acknowledgements). When the notifier exists, delivered ids are routed through finalizer.as_ref().expect(\"Finalizer must have been set up for acknowledgements\"). The invariant: whenever acknowledgements are enabled (so notifier is Some), handle_response is called with a Some finalizer. The panic means notifier and finalizer were built from different assumptions - an internal wiring bug, not a GCP-side condition.","triggerScenarios":"Invoking the streaming-pull loop with acknowledgements enabled but without constructing the Finalizer - custom forks or refactors that create the BatchNotifier path but skip the finalizer setup. Stock Vector derives both from the same source config, so the branch should be unreachable.","commonSituations":"Custom builds between gcp_pubsub acknowledgement refactors; toggling acknowledgement settings in forks without rebuilding the finalizer path.","solutions":["On stock Vector: report with config and version - this is an unreachable-invariant bug worth fixing structurally","Workaround: run the source with acknowledgements disabled if your pipeline semantics allow it","Patch: pass a plain Finalizer (not Option) when acks are on, or derive finalizer and notifier from a single Option so they cannot diverge"],"exampleFix":"// before\nSome(notifier) => {\n    finalizer\n        .as_ref()\n        .expect(\"Finalizer must have been set up for acknowledgements\")\n        .add(ids, notifier);\n}\n\n// after - derive both from one value so they cannot diverge\nmatch (finalizer.as_ref(), notifier) {\n    (Some(finalizer), Some(notifier)) => {\n        finalizer.add(ids, notifier);\n        *pending_acks += 1;\n    }\n    (None, Some(_)) => {\n        error!(message = \"acknowledgements enabled without finalizer; acking directly\");\n        ack_ids.send(ids).await.ok();\n    }\n    (_, None) => { ack_ids.send(ids).await.ok(); }\n}","handlingStrategy":"validation","validationCode":"// make the two pieces impossible to diverge before starting the pull loop:\nlet (finalizer, acks_enabled) = if config acknowledgements {\n    (Some(finalizer), true)\n} else {\n    (None, false)\n};\ndebug_assert_eq!(finalizer.is_some(), acks_enabled, \"finalizer must exist iff acknowledgements are enabled\");","typeGuard":"fn ack_pair(finalizer: &Option<Finalizer>, notifier: &Option<BatchNotifier>) -> bool {\n    finalizer.is_some() || notifier.is_none()\n}","tryCatchPattern":"match (finalizer.as_ref(), notifier) {\n    (Some(f), Some(n)) => { f.add(ids, n); *pending_acks += 1; }\n    (None, Some(_)) => { error!(\"acks enabled without finalizer\"); ack_ids.send(ids).await.ok(); }\n    (_, None) => { ack_ids.send(ids).await.ok(); }\n}","preventionTips":["Derive coupled values (notifier, finalizer) from a single source of truth so they cannot diverge","Add debug assertions on pairing invariants at construction time","Cover acknowledgement-enabled and disabled paths with integration tests"],"tags":["rust","panic","invariant","gcp-pubsub","acknowledgements"],"backgroundTag":"invariant-assertion-failed","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}