What causes "unable to send stream chunk" in risingwavelabs/risingwave?

Inspect the downstream consumer's logs for the root failure that dropped the reader.

unable to send stream chunk

Error message

unable to send stream chunk

What it means

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.

Source

Thrown at src/stream/src/common/log_store_impl/in_mem.rs:300

    ) -> LogStoreResult<()> {
        let init_epoch_tx = self.init_epoch_tx.take().expect("cannot be init for twice");
        self.wait_init_epoch
            .take()
            .expect("cannot be init for in-mem log store")(epoch)
        .await?;
        init_epoch_tx
            .send(epoch.curr)
            .map_err(|_| anyhow!("unable to send init epoch"))?;
        self.curr_epoch = Some(epoch.curr);
        Ok(())
    }

    async fn write_chunk(&mut self, chunk: StreamChunk) -> LogStoreResult<()> {
        self.item_tx
            .send(InMemLogStoreItem::StreamChunk(chunk))
            .instrument_await("in_mem_send_item_chunk")
            .await
            .map_err(|_| anyhow!("unable to send stream chunk"))?;
        Ok(())
    }

    async fn flush_current_epoch(
        &mut self,
        next_epoch: u64,
        options: FlushCurrentEpochOptions,
    ) -> LogStoreResult<LogWriterPostFlushCurrentEpoch<'_>> {
        let is_checkpoint = options.is_checkpoint;
        self.item_tx
            .send(InMemLogStoreItem::Barrier {
                next_epoch,
                options,
            })
            .instrument_await("in_mem_send_item_barrier")
            .await
            .map_err(|_| anyhow!("unable to send barrier"))?;

View on GitHub (pinned to 6469eb736d)

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.

Example fix

// before: reader dropped while writer alive
drop(reader);
writer.write_chunk(chunk).await?;

// after: reader lives as long as the writer
let reader_handle = tokio::spawn(consume(reader));
writer.write_chunk(chunk).await?;

Defensive patterns

Strategy: try-catch

Validate before calling

if !reader_is_alive() { // do not write into a dead pair
    fail_actor();
}

Try / catch

if let Err(e) = writer.write_chunk(chunk).await {
    if e.to_string().contains("unable to send stream chunk") {
        // terminal: reader dropped, stop writing
        fail_actor(e);
    }
    return Err(e);
}

Prevention

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.

When it happens

Trigger: Calling `write_chunk` after the BoundedInMemLogStoreReader was dropped — the downstream executor terminated, panicked, or was cancelled while the writer kept receiving data.

Common situations: 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.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/d8125cff84bff5e7. Report an issue: GitHub.