{"record":{"id":"97f6230eb7e0cbfa","repo":"risingwavelabs/risingwave","slug":"epoch-does-not-match-with-current-epoch","errorCode":null,"errorMessage":"epoch {} does not match with current epoch {}","messagePattern":"epoch (.+?) does not match with current epoch (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/kv_log_store/serde.rs","lineNumber":946,"sourceCode":"        self.row_streams.push(stream.into_future());\n        while let Some((stream_epoch, _)) = self.not_started_streams.last()\n            && *stream_epoch == epoch\n        {\n            let (_, stream) = self.not_started_streams.pop().expect(\"should not be empty\");\n            self.row_streams.push(stream.into_future());\n        }\n        self.stream_state = StreamState::AllConsumingRow { curr_epoch: epoch };\n        Ok(true)\n    }\n\n    fn may_init_epoch(&mut self, epoch: u64) -> LogStoreResult<()> {\n        let prev_epoch = match &self.stream_state {\n            StreamState::Uninitialized => unreachable!(\"should have init\"),\n            StreamState::BarrierEmitted { prev_epoch } => *prev_epoch,\n            StreamState::AllConsumingRow { curr_epoch }\n            | StreamState::BarrierAligning { curr_epoch, .. } => {\n                return if *curr_epoch != epoch {\n                    Err(anyhow!(\n                        \"epoch {} does not match with current epoch {}\",\n                        epoch,\n                        curr_epoch\n                    ))\n                } else {\n                    Ok(())\n                };\n            }\n        };\n\n        if prev_epoch >= epoch {\n            return Err(anyhow!(\n                \"epoch {} should be greater than prev epoch {}\",\n                epoch,\n                prev_epoch\n            ));\n        }\n","sourceCodeStart":928,"sourceCodeEnd":964,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/kv_log_store/serde.rs#L928-L964","documentation":"This error is thrown by the KV log store's epoch validation (`validate_epoch`) when a barrier/write is issued for an epoch that differs from the epoch the stream state machine is currently tracking (`AllConsumingRow` or `BarrierAligning`). It means the log store expected all work for the current epoch to be finished and a matching-epoch operation before emitting a barrier, but got a different epoch. It protects the invariant that barrier epochs strictly follow the epochs of the data rows they cover.","triggerScenarios":"Calling the log store's epoch validation path (e.g. when appending a barrier or updating stream state) while `self.stream_state` is `StreamState::AllConsumingRow { curr_epoch }` or `StreamState::BarrierAligning { curr_epoch, .. }` and the incoming `epoch != curr_epoch`.","commonSituations":"Bugs in stream actor epoch progression (actors consuming rows of epoch N but a barrier for epoch N+1 or N-1 arriving first), recovery/bootstrap replaying epochs out of order, or mixing state from a restarted Hummock epoch assignment with stale in-memory stream state.","solutions":["Check that all actors have finished consuming rows of the current epoch before the barrier for the next epoch is propagated (barrier alignment).","Verify epoch assignment source (meta/Hummock epoch) has not gone backwards; log both epochs at the call site.","If this occurs during recovery, ensure the log store state was correctly restored from the manifest before replaying epochs.","Inspect upstream barrier injection for duplicated or out-of-order `Epoch` values in `Barrier` messages."],"exampleFix":"// before: emitting barrier with a fresh epoch while rows of old epoch are still aligning\nlog_store.append_barrier(barrier.with_epoch(new_epoch)).await?;\n// after: only advance epoch after the stream state confirms all rows of current epoch are consumed\nif matches!(state, StreamState::AllConsumingRow { .. }) {\n    log_store.append_barrier(barrier.with_epoch(new_epoch)).await?;\n}","handlingStrategy":"validation","validationCode":"// Rust: verify epoch matches current stream state before appending a barrier\nfn epoch_matches_state(state: &StreamState, epoch: u64) -> bool {\n    match state {\n        StreamState::AllConsumingRow { curr_epoch }\n        | StreamState::BarrierAligning { curr_epoch, .. } => *curr_epoch == epoch,\n        StreamState::BarrierEmitted { .. } | StreamState::Uninitialized => true,\n    }\n}\nif !epoch_matches_state(&state, epoch) { /* defer or reject barrier */ }","typeGuard":"fn is_row_phase(state: &StreamState) -> bool {\n    matches!(state, StreamState::AllConsumingRow { .. } | StreamState::BarrierAligning { .. })\n}","tryCatchPattern":null,"preventionTips":["Advance epochs only after barrier alignment confirms all actors finished the current epoch.","Take epochs from a single authoritative source (meta node epoch provider).","Log previous and new epochs on every barrier to spot out-of-order flow early."],"tags":["streaming","epoch","barrier","state-machine"],"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-14T11:17:12.474Z"}