What causes "unable to send barrier" in risingwavelabs/risingwave?

Look for an earlier downstream executor failure in the logs; that is the root cause, this error is secondary.

unable to send barrier

Error message

unable to send barrier

What it means

flush_current_epoch sends an InMemLogStoreItem::Barrier to the reader over the shared mpsc channel. This error means the send failed because the reader's receiver was dropped: the barrier cannot be acknowledged and the epoch cannot be checkpointed. Like the other in-mem log store send errors, it indicates the consuming side of the pair terminated unexpectedly.

Source

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

            .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"))?;

        let prev_epoch = self
            .curr_epoch
            .replace(next_epoch)
            .expect("should have epoch");

        if is_checkpoint {
            let truncated_epoch = self
                .truncated_epoch_rx
                .recv()
                .instrument_await("in_mem_recv_truncated_epoch")
                .await
                .ok_or_else(|| anyhow!("cannot get truncated epoch"))?;
            assert_eq!(truncated_epoch, prev_epoch);
        }

        Ok(LogWriterPostFlushCurrentEpoch::new(move || {
            async move { Ok(()) }.boxed()

View on GitHub (pinned to 6469eb736d)

Solutions

Look for an earlier downstream executor failure in the logs; that is the root cause, this error is secondary.
Let the actor fail and rely on recovery rather than retrying the barrier send.
Verify writer/reader pairing: the reader must be owned by a live consuming task during flush.
Check the await-tree trace (`in_mem_send_item_barrier`) to see where the send blocked before failing.

Defensive patterns

Strategy: try-catch

Try / catch

if let Err(e) = writer.flush_current_epoch(epoch, opts).await {
    if e.to_string().contains("unable to send barrier") {
        // reader gone; treat epoch as not checkpointed and fail actor
        fail_actor(e);
    }
    return Err(e);
}

Prevention

Monitor downstream actor health before barrier flush.
Ensure barrier handling and reader consumption run in paired tasks.
Use await-tree traces to catch blocked barrier sends early.

When it happens

Trigger: Calling `flush_current_epoch` when the paired reader has been dropped — the consumer actor exited, panicked, or was cancelled before the barrier was sent.

Common situations: Barrier completion failing during actor failure recovery; downstream executor crash mid-epoch; cancellation during pause/migration that dropped the reader while the writer flushed.

Related errors

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