{"record":{"id":"5df1908699da157b","repo":"risingwavelabs/risingwave","slug":"cannot-get-truncated-epoch","errorCode":null,"errorMessage":"cannot get truncated epoch","messagePattern":"cannot get truncated epoch","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/in_mem.rs","lineNumber":330,"sourceCode":"                next_epoch,\n                options,\n            })\n            .instrument_await(\"in_mem_send_item_barrier\")\n            .await\n            .map_err(|_| anyhow!(\"unable to send barrier\"))?;\n\n        let prev_epoch = self\n            .curr_epoch\n            .replace(next_epoch)\n            .expect(\"should have epoch\");\n\n        if is_checkpoint {\n            let truncated_epoch = self\n                .truncated_epoch_rx\n                .recv()\n                .instrument_await(\"in_mem_recv_truncated_epoch\")\n                .await\n                .ok_or_else(|| anyhow!(\"cannot get truncated epoch\"))?;\n            assert_eq!(truncated_epoch, prev_epoch);\n        }\n\n        Ok(LogWriterPostFlushCurrentEpoch::new(move || {\n            async move { Ok(()) }.boxed()\n        }))\n    }\n\n    fn pause(&mut self) -> LogStoreResult<()> {\n        // no-op when decouple is not enabled\n        Ok(())\n    }\n\n    fn resume(&mut self) -> LogStoreResult<()> {\n        // no-op when decouple is not enabled\n        Ok(())\n    }\n}","sourceCodeStart":312,"sourceCodeEnd":348,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/in_mem.rs#L312-L348","documentation":"After a checkpoint barrier, the writer waits for the reader to report the truncated epoch back over an unbounded channel. `recv()` returning None means every `truncated_epoch_tx` sender was dropped, i.e. the reader is gone, so the writer can never learn whether the epoch was consumed. The error therefore signals a dead reader at checkpoint time.","triggerScenarios":"Calling `flush_current_epoch` with a checkpoint (is_checkpoint=true) after the reader was dropped, so `truncated_epoch_rx.recv()` returns None instead of the expected prev_epoch.","commonSituations":"Reader actor crashed or was cancelled before it could ack the previous checkpoint epoch; job restart/scale-in terminating the consumer; test harness dropping the reader early.","solutions":["Find the upstream failure that terminated the reader task; this error is a consequence.","Ensure the reader runs a consume loop that calls truncate after each checkpoint epoch so the ack is sent.","Treat as terminal: fail the writing actor and let recovery rebuild the log store pair.","In tests, keep the truncated_epoch_tx alive (keep the reader instance) until flush completes."],"exampleFix":"// before: reader dropped before ack\nlet (mut writer, reader) = factory.create();\ndrop(reader);\nwriter.flush_current_epoch(epoch, opts).await?;\n\n// after: reader alive and truncating\nlet mut reader = reader;\nreader.truncate(offset)?;\nwriter.flush_current_epoch(epoch, opts).await?;","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"if let Err(e) = writer.flush_current_epoch(epoch, opts).await {\n    if e.to_string().contains(\"cannot get truncated epoch\") {\n        // reader dropped before acking; fail actor for recovery\n        fail_actor(e);\n    }\n    return Err(e);\n}","preventionTips":["Reader must consume and truncate each checkpoint epoch to send its ack.","Never drop the reader before the writer finishes flushing the current epoch.","Add supervision/restart on the reader task to avoid silent drops."],"tags":["channel-closed","checkpoint","log-store","streaming"],"backgroundTag":"broken-pipe","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"}