{"record":{"id":"809ef9bb4fb93802","repo":"risingwavelabs/risingwave","slug":"unable-to-send-sealed-epoch","errorCode":null,"errorMessage":"unable to send sealed epoch","messagePattern":"unable to send sealed epoch","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/in_mem.rs","lineNumber":262,"sourceCode":"            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\"))?;\n        }\n        self.truncate_offset = offset;\n        Ok(())\n    }\n\n    async fn rewind(&mut self) -> LogStoreResult<()> {\n        Err(anyhow!(\"should not call rewind on it\"))\n    }\n\n    async fn start_from(&mut self, _start_offset: Option<u64>) -> LogStoreResult<()> {\n        Ok(())\n    }\n}\n\nimpl LogWriter for BoundedInMemLogStoreWriter {\n    async fn init(\n        &mut self,\n        epoch: EpochPair,","sourceCodeStart":244,"sourceCodeEnd":280,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/in_mem.rs#L244-L280","documentation":"While truncating an epoch, the in-memory log store tried to notify the sealed-epoch watcher via a broadcast channel but all receivers were dropped, so the send failed. This indicates the sealed-epoch consumer side is gone.","triggerScenarios":"truncate moves epoch progress past a sealed epoch and calls truncated_epoch_tx.send(sealed_epoch) when no receiver is alive.","commonSituations":"The sealed-epoch reader task panicked or was dropped before truncation, shutdown races between the log store and its consumers, or incorrect initialization dropping the receiver.","solutions":["Check the receiver task for panics or early termination in logs","Ensure the sealed-epoch consumer is spawned and alive before truncation happens","Fix shutdown ordering so the log store is not truncating after consumers exit"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"match log_store.truncate(offset).await {\n    Err(e) if e.to_string().contains(\"unable to send sealed epoch\") => {\n        tracing::warn!(\"sealed-epoch receiver gone; check consumer task\");\n        // restart consumer or surface shutdown\n    }\n    other => other?,\n}","preventionTips":["Keep the sealed-epoch receiver task alive across the log store's lifetime","Order shutdown: stop truncation before dropping consumers","Watch consumer task panics in monitoring"],"tags":["stream","log-store","channel"],"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"}