{"record":{"id":"3c93eecbc44081e3","repo":"risingwavelabs/risingwave","slug":"truncate-at-but-latest-offset-is","errorCode":null,"errorMessage":"truncate at {:?} but latest offset is {:?}","messagePattern":"truncate at (.+?) but latest offset is (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/in_mem.rs","lineNumber":244,"sourceCode":"                None => Err(anyhow!(\"end of log stream\")),\n            },\n            AwaitingTruncate { .. } => std::future::pending().await,\n        }\n    }\n\n    fn truncate(&mut self, offset: TruncateOffset) -> LogStoreResult<()> {\n        // check the truncate offset is higher than prev truncate offset\n        if self.truncate_offset >= offset {\n            return Err(anyhow!(\n                \"truncate offset {:?} but prev truncate offset is {:?}\",\n                offset,\n                self.truncate_offset\n            ));\n        }\n\n        // check the truncate offset does not exceed the latest possible offset\n        if offset > self.latest_offset {\n            return Err(anyhow!(\n                \"truncate at {:?} but latest offset is {:?}\",\n                offset,\n                self.latest_offset\n            ));\n        }\n\n        if let AwaitingTruncate {\n            sealed_epoch,\n            next_epoch,\n        } = &self.epoch_progress\n            && let TruncateOffset::Barrier { epoch } = offset\n            && epoch == *sealed_epoch\n        {\n            let sealed_epoch = *sealed_epoch;\n            self.epoch_progress = Consuming(*next_epoch);\n            self.truncated_epoch_tx\n                .send(sealed_epoch)\n                .map_err(|_| anyhow!(\"unable to send sealed epoch\"))?;","sourceCodeStart":226,"sourceCodeEnd":262,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/in_mem.rs#L226-L262","documentation":"Guard in the in-memory log store's truncate: the requested truncate offset exceeds the latest written offset in the buffer. One cannot truncate data that has not been written yet, so out-of-order or over-advancing truncation is rejected with this error.","triggerScenarios":"truncate(offset) invoked when offset > self.latest_offset, e.g. a consumer computing a future offset or using a stale/garbage offset value.","commonSituations":"Bugs in offset bookkeeping (u64 underflow/overflow), mismatched offset domains between log store versions, or corrupted checkpoint metadata supplying bad offsets.","solutions":["Clamp the truncate offset to latest_offset before calling truncate","Verify where the offset came from (checkpoint, watermark) for corruption or version skew","Fix offset computation logic that predicts beyond written entries"],"exampleFix":"// before\nlog_store.truncate(watermark).await?;\n// after\nlog_store.truncate(watermark.min(latest_offset)).await?;","handlingStrategy":"validation","validationCode":"fn sanitize_truncate_offset(offset: u64, latest_offset: u64) -> u64 {\n    offset.min(latest_offset)\n}","typeGuard":null,"tryCatchPattern":"let safe = offset.min(latest_offset);\nif safe > current_truncate_offset {\n    log_store.truncate(safe).await?;\n}","preventionTips":["Clamp offsets against latest_offset before truncating","Audit offset provenance (checkpoints, watermarks) for corruption","Add unit tests for offset underflow/overflow paths"],"tags":["stream","log-store","truncate"],"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-14T16:17:12.679Z"}