{"record":{"id":"8dc027bd40f639dc","repo":"risingwavelabs/risingwave","slug":"invalid-commit-epoch-prev-epoch","errorCode":null,"errorMessage":"invalid commit epoch {}, prev_epoch {}","messagePattern":"invalid commit epoch (.+?), prev_epoch (.+?)","errorType":"validation","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/handle.rs","lineNumber":130,"sourceCode":"\n    pub(super) fn poll_next_request(\n        &mut self,\n        cx: &mut Context<'_>,\n    ) -> Poll<anyhow::Result<coordinate_request::Msg>> {\n        let result = try {\n            let request = ready!(self.request_stream.try_poll_next_unpin(cx))\n                .ok_or_else(|| anyhow!(\"end of request stream\"))?\n                .map_err(anyhow::Error::from)?;\n            let request = request.msg.ok_or_else(|| anyhow!(\"None msg in request\"))?;\n            match &request {\n                coordinate_request::Msg::StartRequest(_)\n                | coordinate_request::Msg::Stop(_)\n                | coordinate_request::Msg::AlignInitialEpochRequest(_) => {}\n                coordinate_request::Msg::CommitRequest(request) => {\n                    if let Some(prev_epoch) = self.prev_epoch\n                        && request.epoch < prev_epoch\n                    {\n                        return Poll::Ready(Err(anyhow!(\n                            \"invalid commit epoch {}, prev_epoch {}\",\n                            request.epoch,\n                            prev_epoch\n                        )));\n                    }\n                    if request.metadata.is_none() {\n                        return Poll::Ready(Err(anyhow!(\"empty commit metadata\")));\n                    };\n                    self.prev_epoch = Some(request.epoch);\n                }\n                coordinate_request::Msg::UpdateVnodeRequest(request) => {\n                    let bitmap = Bitmap::from(\n                        request\n                            .vnode_bitmap\n                            .as_ref()\n                            .ok_or_else(|| anyhow!(\"empty vnode bitmap\"))?,\n                    );\n                    self.vnode_bitmap = bitmap;","sourceCodeStart":112,"sourceCodeEnd":148,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/handle.rs#L112-L148","documentation":"The coordinator tracks the previous committed epoch from `CommitRequest`s and rejects a commit whose epoch is lower than the previously seen one. Since commits must be monotonically non-decreasing in epoch, receiving an older epoch indicates out-of-order or duplicated commit requests. This is a strict protocol invariant of the two-phase sink commit protocol.","triggerScenarios":"A `CommitRequest` arrives whose `epoch` is strictly less than `self.prev_epoch` recorded from an earlier commit on the same coordination stream.","commonSituations":"Sink executor retry logic replays an old commit after a partial failure; epoch alignment (`AlignInitialEpochRequest`) done incorrectly; clock/state skew after failover where a stale coordinator state persists.","solutions":["Verify the sink executor sends commits in strictly increasing epoch order and does not replay stale commits after reconnect.","Ensure `AlignInitialEpochRequest` is sent when a new coordinator session starts so `prev_epoch` is seeded correctly.","Check Hummock/epoch manager for epoch regression in the executor; fix the source of the stale epoch.","If triggered by duplicated retries, make commit requests idempotent (skip re-committing an already committed epoch) on the executor side."],"exampleFix":"// executor side: skip stale commits\nif epoch <= last_committed_epoch { return Ok(()); } // instead of sending CommitRequest{epoch}","handlingStrategy":"validation","validationCode":"// Executor side: enforce monotonic epochs before committing\nif epoch <= last_committed_epoch {\n    tracing::debug!(epoch, \"skipping stale commit\");\n    return Ok(());\n}","typeGuard":null,"tryCatchPattern":"if let Err(e) = coordinator_next_request() {\n    if e.to_string().contains(\"invalid commit epoch\") {\n        tracing::error!(\"epoch regression detected; realign initial epoch and reconnect\");\n    }\n}","preventionTips":["Send commit requests strictly in increasing epoch order","Always send AlignInitialEpochRequest when starting a new coordination session","Make commit retries idempotent by tracking the last committed epoch executor-side"],"tags":["rust","meta","epoch","sink-commit-protocol","invariant-violation"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}