{"record":{"id":"e7e53974191b08ec","repo":"risingwavelabs/risingwave","slug":"infinite","errorCode":null,"errorMessage":"infinite","messagePattern":"infinite","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":242,"sourceCode":"            self.prepared_epochs\n                .push_back((epoch, metadata, schema_change));\n        }\n    }\n\n    async fn ack_committed(&mut self, epoch: u64) -> anyhow::Result<()> {\n        self.backoff_state = None;\n        let (last_epoch, _, _) = self.prepared_epochs.pop_front().expect(\"non-empty\");\n        assert_eq!(last_epoch, epoch);\n\n        commit_and_prune_epoch(&self.db, self.sink_id, epoch, self.last_committed_epoch).await?;\n        self.last_committed_epoch = Some(epoch);\n        Ok(())\n    }\n\n    fn failed_committed(&mut self, epoch: u64, err: SinkError) {\n        assert_eq!(self.prepared_epochs.front().expect(\"non-empty\").0, epoch,);\n        if let Some((prev_fut, strategy)) = &mut self.backoff_state {\n            let new_fut = strategy.next().expect(\"infinite\");\n            *prev_fut = new_fut;\n        } else {\n            let mut strategy = Self::get_retry_backoff_strategy();\n            let backoff_fut = strategy.next().expect(\"infinite\");\n            self.backoff_state = Some((backoff_fut, strategy));\n        }\n        tracing::error!(\n            error = %err.as_report(),\n            %self.sink_id,\n            \"failed to commit epoch {}, Retrying after backoff\",\n            epoch,\n        );\n    }\n\n    fn is_empty(&self) -> bool {\n        self.pending_epochs.is_empty() && self.prepared_epochs.is_empty()\n    }\n","sourceCodeStart":224,"sourceCodeEnd":260,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L224-L260","documentation":"This panic fires in `failed_committed` when the sink coordinator needs a backoff delay before retrying a failed epoch commit, and `strategy.next()` on the retry backoff strategy returns `None`. `Backoff::next()` only returns `None` once the iterator is exhausted, which for a properly configured `Backoff` (e.g. `Backoff::exponential` with `max_attempts` unset / infinite) never happens. The `.expect(\"infinite\")` encodes the invariant that the meta-node must retry failed commits forever, so exhausting the strategy is an internal bug in how the backoff strategy was constructed.","triggerScenarios":"A sink commit of a prepared epoch fails with a `SinkError`, `failed_committed` is invoked, and the existing or freshly created `Backoff` strategy yields `None` from `next()` — only possible if the strategy was built with a finite attempt count (e.g. `max_attempts` set) or was fully consumed across retries.","commonSituations":"Someone changed `get_retry_backoff_strategy` to a bounded backoff (e.g. for testing) and shipped it; or a finite `max_attempts` was configured while downstream code still assumes endless retries; persistent sink (e.g. Kafka) outages that drive the retry loop through more attempts than the bound allows.","solutions":["Ensure `get_retry_backoff_strategy` builds an unbounded backoff (no `max_attempts`/finite attempt limit) so `next()` never returns `None`.","Check for local modifications or feature flags that swap the strategy for a bounded one in tests or config.","If retries should be bounded, replace the `.expect(\"infinite\")` handling with graceful abort/failover logic instead of panicking.","While the panic persists, fix the underlying `SinkError` (check sink connectivity) and restart the meta node to reset `backoff_state`."],"exampleFix":"// before\nlet mut strategy = Backoff::exponential(Duration::from_millis(100)).max_attempts(10);\nlet backoff_fut = strategy.next().expect(\"infinite\");\n\n// after\nlet mut strategy = Backoff::exponential(Duration::from_millis(100)); // unbounded\nlet backoff_fut = strategy.next().expect(\"infinite\");","handlingStrategy":"validation","validationCode":"// Before deploying, assert the backoff strategy never exhausts\nlet mut s = CoordinatorWorker::get_retry_backoff_strategy();\nfor _ in 0..1_000_000 { assert!(s.next().is_some(), \"backoff must be infinite\"); }","typeGuard":null,"tryCatchPattern":"// Cannot be caught at runtime (panic); guard construction:\nfn get_retry_backoff_strategy() -> Backoff {\n    Backoff::exponential(Duration::from_millis(100)) // no max_attempts\n}","preventionTips":["Never set max_attempts on the commit-retry backoff","Add a unit test asserting next() never returns None","Review test-only backoff configurations before merging"],"tags":["retry","backoff","panic","sink-coordination","rust"],"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-14T11:17:12.474Z"}