{"record":{"id":"bf05c6283ccf4e3a","repo":"risingwavelabs/risingwave","slug":"next-offset-should-be-later-than-current-offs","errorCode":null,"errorMessage":"next offset {:?} should be later than current offset {:?}","messagePattern":"next offset (.+?) should be later than current offset (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/log_store.rs","lineNumber":79,"sourceCode":"}\n\nimpl TruncateOffset {\n    pub fn next_chunk_id(&self) -> ChunkId {\n        match self {\n            TruncateOffset::Chunk { chunk_id, .. } => chunk_id + 1,\n            TruncateOffset::Barrier { .. } => 0,\n        }\n    }\n\n    pub fn epoch(&self) -> u64 {\n        match self {\n            TruncateOffset::Chunk { epoch, .. } | TruncateOffset::Barrier { epoch } => *epoch,\n        }\n    }\n\n    pub fn check_next_offset(&self, next_offset: TruncateOffset) -> LogStoreResult<()> {\n        if *self >= next_offset {\n            bail!(\n                \"next offset {:?} should be later than current offset {:?}\",\n                next_offset,\n                self\n            )\n        } else {\n            Ok(())\n        }\n    }\n\n    pub fn check_next_item_epoch(&self, epoch: u64) -> LogStoreResult<()> {\n        match self {\n            TruncateOffset::Chunk {\n                epoch: offset_epoch,\n                ..\n            } => {\n                if epoch != *offset_epoch {\n                    bail!(\n                        \"new item epoch {} does not match current chunk offset epoch {}\",","sourceCodeStart":61,"sourceCodeEnd":97,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/log_store.rs#L61-L97","documentation":"LogStore ordering invariant check: `check_next_offset` on the current truncate offset rejects any proposed next truncate offset that is not strictly later than the current one. It guards the sink log store against rewinding or replaying offsets, which would corrupt the sink's at-least-once/exactly-once delivery bookkeeping.","triggerScenarios":"Calling `LogStore::check_next_offset(next_offset)` with a `TruncateOffset` whose epoch is <= the currently stored offset's epoch, e.g. a sink writer producing stale chunk/barrier offsets after a failure or epoch regression.","commonSituations":"A sink implementation truncating with an offset captured before a barrier; epoch values reused after recovery; a sink connector that reorders epochs under concurrent writers.","solutions":["Fix the sink writer to only advance the truncate offset (next epoch must be strictly greater)","Check for stale offsets cached across barrier/recovery and refresh them before calling check_next_offset","Ensure only one writer drives the log store for a given sink"],"exampleFix":"// before\nstore.check_next_offset(TruncateOffset::Barrier { epoch: old_epoch });\n// after\nif epoch > current_epoch {\n    store.check_next_offset(TruncateOffset::Barrier { epoch });\n}","handlingStrategy":"validation","validationCode":"fn can_advance(current: &TruncateOffset, next: &TruncateOffset) -> bool {\n    next.epoch_value() > current.epoch_value()\n}","typeGuard":"fn is_later(next: &TruncateOffset, current: &TruncateOffset) -> bool {\n    match (next, current) {\n        (TruncateOffset::Chunk { epoch: n, .. } | TruncateOffset::Barrier { epoch: n },\n         TruncateOffset::Chunk { epoch: c, .. } | TruncateOffset::Barrier { epoch: c }) => n > c,\n    }\n}","tryCatchPattern":"match store.check_next_offset(next) {\n    Ok(()) => store.advance(next),\n    Err(e) => { warn!(\"stale offset skipped: {e}\"); /* refresh offset from barrier */ }\n}","preventionTips":["Stamp offsets from a single monotonic epoch source","Refresh cached offsets after every barrier and recovery","Avoid multiple writers advancing the same log store"],"tags":["rust","streaming","offset-ordering"],"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"}