{"record":{"id":"d8125cff84bff5e7","repo":"risingwavelabs/risingwave","slug":"unable-to-send-stream-chunk","errorCode":null,"errorMessage":"unable to send stream chunk","messagePattern":"unable to send stream chunk","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/in_mem.rs","lineNumber":300,"sourceCode":"    ) -> 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<'_>> {\n        let is_checkpoint = options.is_checkpoint;\n        self.item_tx\n            .send(InMemLogStoreItem::Barrier {\n                next_epoch,\n                options,\n            })\n            .instrument_await(\"in_mem_send_item_barrier\")\n            .await\n            .map_err(|_| anyhow!(\"unable to send barrier\"))?;\n","sourceCodeStart":282,"sourceCodeEnd":318,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/in_mem.rs#L282-L318","documentation":"write_chunk forwards a StreamChunk to the paired reader through a bounded mpsc channel. The `.await` only fails with this error when the reader's receiver has been dropped, i.e. the consumer side of the in-memory log store no longer exists. It signals that the writing actor is writing to a dead log store pair and its output can no longer be delivered.","triggerScenarios":"Calling `write_chunk` after the BoundedInMemLogStoreReader was dropped — the downstream executor terminated, panicked, or was cancelled while the writer kept receiving data.","commonSituations":"Downstream actor crash (e.g. a sink error) that killed the reader task; job cancellation racing with incoming data; mis-paired writer/reader lifetimes in custom executor code or tests.","solutions":["Inspect the downstream consumer's logs for the root failure that dropped the reader.","Fail fast: treat this as terminal for the actor instead of retrying writes.","Ensure the reader task is spawned and kept alive for the entire lifetime of the writer.","In tests, keep the reader handle alive (e.g. hold it in a spawned task) while writing."],"exampleFix":"// before: reader dropped while writer alive\ndrop(reader);\nwriter.write_chunk(chunk).await?;\n\n// after: reader lives as long as the writer\nlet reader_handle = tokio::spawn(consume(reader));\nwriter.write_chunk(chunk).await?;","handlingStrategy":"try-catch","validationCode":"if !reader_is_alive() { // do not write into a dead pair\n    fail_actor();\n}","typeGuard":null,"tryCatchPattern":"if let Err(e) = writer.write_chunk(chunk).await {\n    if e.to_string().contains(\"unable to send stream chunk\") {\n        // terminal: reader dropped, stop writing\n        fail_actor(e);\n    }\n    return Err(e);\n}","preventionTips":["Keep the reader task alive for the writer's whole lifetime.","Propagate downstream failures upstream promptly.","Avoid dropping reader handles in tests while writes are in flight."],"tags":["channel-closed","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"}