{"record":{"id":"39e6f01c77a28cde","repo":"risingwavelabs/risingwave","slug":"non-empty","errorCode":null,"errorMessage":"non-empty","messagePattern":"non-empty","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":179,"sourceCode":"            .map(jitter)\n            .map(|delay| Box::pin(tokio::time::sleep(delay)))\n    }\n\n    async fn next_to_commit(\n        &mut self,\n    ) -> anyhow::Result<(u64, Option<Vec<u8>>, Option<PbSinkSchemaChange>)> {\n        loop {\n            let wait_backoff = async {\n                if self.prepared_epochs.is_empty() {\n                    pending::<()>().await;\n                } else if let Some((backoff_fut, _)) = &mut self.backoff_state {\n                    backoff_fut.await;\n                }\n            };\n\n            select! {\n                _ = wait_backoff => {\n                    let item = self.prepared_epochs.front().cloned().expect(\"non-empty\");\n                    return Ok(item);\n                }\n\n                recv_epoch = self.job_committed_epoch_rx.recv() => {\n                    let Some(recv_epoch) = recv_epoch else {\n                        return Err(anyhow!(\n                            \"Hummock committed epoch sender closed unexpectedly\"\n                        ));\n                    };\n                    self.curr_hummock_committed_epoch = recv_epoch;\n                    while let Some((epoch, metadata, schema_change)) = self.pending_epochs.pop_front_if(|(epoch, _, _)| *epoch <= recv_epoch) {\n                        if let Some((last_epoch, _, _)) = self.prepared_epochs.back() {\n                            assert!(epoch > *last_epoch, \"prepared epochs must be in increasing order\");\n                        }\n                        self.prepared_epochs.push_back((epoch, metadata, schema_change));\n                    }\n                }\n            }","sourceCodeStart":161,"sourceCodeEnd":197,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L161-L197","documentation":"An `.expect(\"non-empty\")` panic in `next_to_commit`: after the backoff wait resolves, the code assumes `prepared_epochs` is non-empty and takes its front element. The invariant is that the backoff timer is only armed when a prepared epoch exists; if it fires after that epoch was removed, the invariant is broken.","triggerScenarios":"The backoff future fires after `prepared_epochs` was drained by `ack_committed`/`failed_committed` or by the `job_committed_epoch_rx` branch — a race between the select arms or a bug leaving the backoff state armed for a consumed epoch.","commonSituations":"High-frequency commit/ack churn where backoff timers overlap epoch transitions; meta-node task scheduling delays letting a stale timer win the select; bugs in resetting `backoff_state` when epochs are consumed.","solutions":["Re-check `prepared_epochs` after the backoff wait and loop/re-arm instead of blindly taking the front element.","Ensure `ack_committed`/`failed_committed` clear or recompute `backoff_state` whenever `prepared_epochs` changes.","Inspect logs for interleaved ack/commit of epochs racing with backoff waits and reproduce to fix the race.","Replace `.expect(\"non-empty\")` with an early `return`/continue when the queue is empty."],"exampleFix":"// before\nlet item = self.prepared_epochs.front().cloned().expect(\"non-empty\");\nreturn Ok(item);\n// after\nmatch self.prepared_epochs.front().cloned() {\n    Some(item) => return Ok(item),\n    None => { self.backoff_state = None; continue; } // timer fired for a consumed epoch\n}","handlingStrategy":"validation","validationCode":"// check queue before acting on the backoff timer\nif self.prepared_epochs.is_empty() { self.backoff_state = None; return Ok(()); }","typeGuard":"fn has_prepared(q: &VecDeque<(u64, _, _)>) -> bool { !q.is_empty() }","tryCatchPattern":"// panic-based: cannot catch; restructure the select so the backoff branch re-checks the queue\nif self.prepared_epochs.front().is_none() { continue; }","preventionTips":["Clear or recompute backoff_state whenever prepared_epochs is mutated","Re-validate invariants after each select arm instead of assuming state from before the wait","Add unit tests interleaving backoff timers with acks/failures"],"tags":["meta","sink","panic","race"],"backgroundTag":"internal-invariant-violation","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"}