{"record":{"id":"de3dd5162bc0d05c","repo":"risingwavelabs/risingwave","slug":"truncate-at-a-later-offset-than-the-current-l","errorCode":null,"errorMessage":"truncate at a later offset {:?} than the current latest offset {:?}","messagePattern":"truncate at a later offset (.+?) than the current latest offset (.+?)","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/kv_log_store/reader.rs","lineNumber":519,"sourceCode":"                    item_epoch\n                );\n                self.latest_offset = Some(TruncateOffset::Barrier { epoch: item_epoch });\n                (\n                    item_epoch,\n                    LogStoreReadItem::Barrier {\n                        is_checkpoint,\n                        new_vnode_bitmap: None,\n                        is_stop,\n                        schema_change,\n                    },\n                )\n            }\n        })\n    }\n\n    fn truncate(&mut self, offset: TruncateOffset) -> LogStoreResult<()> {\n        if offset > self.latest_offset.expect(\"should exist before truncation\") {\n            return Err(anyhow!(\n                \"truncate at a later offset {:?} than the current latest offset {:?}\",\n                offset,\n                self.latest_offset\n            ));\n        }\n        if offset.epoch() >= self.first_write_epoch.expect(\"should have init\") {\n            if let Some(truncate_offset) = &self.truncate_offset\n                && offset <= *truncate_offset\n            {\n                return Err(anyhow!(\n                    \"truncate offset {:?} earlier than prev truncate offset {:?}\",\n                    offset,\n                    truncate_offset\n                ));\n            }\n            self.rx.truncate_buffer(offset);\n            self.truncate_offset = Some(offset);\n        } else {","sourceCodeStart":501,"sourceCodeEnd":537,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/kv_log_store/reader.rs#L501-L537","documentation":"KvLogStoreReader::truncate refuses to truncate at an offset later than the reader's `latest_offset` — the reader has not consumed that far, so a later truncation would discard data the reader still needs. `latest_offset` is expected to be Some (it panics with 'should exist before truncation' otherwise), and the error here covers the out-of-range case. This guards the invariant that truncation only moves forward within consumed data.","triggerScenarios":"Calling `truncate(offset)` with an offset whose epoch (or position within the epoch) exceeds what the reader has already emitted: e.g. truncating based on a writer-side sealed epoch while the reader lags, or a stale reader instance receiving a new truncate request after failover.","commonSituations":"Reader lagging behind barrier truncation requests; failover where a new owner truncates an old reader; tests or custom callers passing a TruncateOffset derived from the wrong reader's progress.","solutions":["Only truncate with offsets derived from this reader's own emitted progress, not the writer's.","Ensure the reader has consumed up to the target epoch (call next_item until latest_offset passes it) before truncating.","Check for duplicate/overlapping truncate calls after failover; discard stale reader instances.","Add a debug log comparing the requested offset to latest_offset to catch ordering bugs."],"exampleFix":"// before: truncating beyond consumed progress\nlet offset = TruncateOffset { epoch: barrier_epoch, .. };\nreader.truncate(offset)?;\n\n// after: truncate only up to the reader's latest consumed offset\nlet offset = reader.latest_offset.min(requested);\nreader.truncate(offset)?;","handlingStrategy":"validation","validationCode":"// only truncate offsets this reader has consumed\nif offset > reader.latest_offset {\n    // clamp or skip instead of calling truncate\n    return Ok(());\n}\nreader.truncate(offset)?;","typeGuard":"fn truncatable(offset: TruncateOffset, latest: Option<TruncateOffset>) -> bool {\n    matches!(latest, Some(l) if offset <= l)\n}","tryCatchPattern":"match reader.truncate(offset) {\n    Err(e) if e.to_string().contains(\"truncate at a later offset\") => {\n        // reader lags or stale instance; re-derive offset from reader progress\n        let off = reader.latest_offset();\n        reader.truncate(off)?;\n    }\n    r => r,\n}","preventionTips":["Derive truncate offsets from the reader's own latest_offset, never from writer-side epochs.","Consume the reader up to the target epoch before truncating.","Discard stale reader instances after failover to avoid out-of-order truncates."],"tags":["truncation","offset","invariant","log-store"],"backgroundTag":"value-out-of-range","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"}