{"record":{"id":"b203d56d4ddf65d3","repo":"risingwavelabs/risingwave","slug":"unable-to-send-init-epoch","errorCode":null,"errorMessage":"unable to send init epoch","messagePattern":"unable to send init epoch","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/in_mem.rs","lineNumber":290,"sourceCode":"    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,\n        _pause_read_on_bootstrap: bool,\n    ) -> LogStoreResult<()> {\n        let init_epoch_tx = self.init_epoch_tx.take().expect(\"cannot be init for twice\");\n        self.wait_init_epoch\n            .take()\n            .expect(\"cannot be init for in-mem log store\")(epoch)\n        .await?;\n        init_epoch_tx\n            .send(epoch.curr)\n            .map_err(|_| anyhow!(\"unable to send init epoch\"))?;\n        self.curr_epoch = Some(epoch.curr);\n        Ok(())\n    }\n\n    async fn write_chunk(&mut self, chunk: StreamChunk) -> LogStoreResult<()> {\n        self.item_tx\n            .send(InMemLogStoreItem::StreamChunk(chunk))\n            .instrument_await(\"in_mem_send_item_chunk\")\n            .await\n            .map_err(|_| anyhow!(\"unable to send stream chunk\"))?;\n        Ok(())\n    }\n\n    async fn flush_current_epoch(\n        &mut self,\n        next_epoch: u64,\n        options: FlushCurrentEpochOptions,\n    ) -> LogStoreResult<LogWriterPostFlushCurrentEpoch<'_>> {","sourceCodeStart":272,"sourceCodeEnd":308,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/in_mem.rs#L272-L308","documentation":"BoundedInMemLogStoreWriter::init hands the first checkpoint epoch to the paired reader over a oneshot channel. This error is raised when `init_epoch_tx.send` fails, which only happens if the receiving reader side was already dropped. It means the writer and reader of the in-memory log store are no longer paired: the downstream actor/consumer that owns the reader has terminated before the writer could initialize.","triggerScenarios":"Calling `init()` on a BoundedInMemLogStoreWriter whose paired BoundedInMemLogStoreReader was already dropped (reader's `init_epoch_rx` gone). Typically the downstream stream executor finished, panicked, or was cancelled before the writer's init ran.","commonSituations":"Actor failure upstream of the writer: the reader's task exited due to an earlier error, a cancellation during streaming job migration or scaling, or a panic in the consumer that dropped the receiver while the writer still runs.","solutions":["Check the logs of the downstream executor/actor that owns the reader for an earlier panic or failure that dropped it.","Ensure the writer and reader are created and driven by the same lifecycle (factory pairing) so neither outlives the other.","Verify no code path calls writer `init()` after the streaming task has been cancelled; treat this error as terminal for the actor.","Reproduce with a minimal test that drops the reader before writer init to confirm the pairing bug."],"exampleFix":"// before: reader dropped, writer init panics/fails later\nlet (writer, reader) = factory.build();\ndrop(reader);\nwriter.init(epoch).await?;\n\n// after: keep reader alive in its consumer task before writer init\nlet (writer, reader) = factory.build();\ntokio::spawn(async move { let mut r = reader; r.init().await; /* consume */ });\nwriter.init(epoch).await?;","handlingStrategy":"try-catch","validationCode":"if reader_dropped_or_terminated() { fail_actor(); } // check pairing before writer.init","typeGuard":"fn writer_init_ok(init_epoch_tx: &Option<oneshot::Sender<u64>>) -> bool { init_epoch_tx.is_some() }","tryCatchPattern":"match writer.init(epoch).await {\n    Ok(()) => {},\n    Err(e) if e.to_string().contains(\"unable to send init epoch\") => {\n        // reader side gone; fail actor and rely on recovery\n        fail_actor(e);\n    },\n    Err(e) => return Err(e),\n}","preventionTips":["Always pair writer and reader lifetimes within the same actor/factory scope.","Fail-fast downstream errors so the writer does not keep running after the reader dies.","Add await-tree/metrics alerts on reader termination ordering."],"tags":["channel-closed","log-store","streaming","lifecycle"],"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"}