risingwavelabs/risingwave · error

unable to send init epoch

Error message

unable to send init epoch

What it means

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.

Source

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

    async fn start_from(&mut self, _start_offset: Option<u64>) -> LogStoreResult<()> {
        Ok(())
    }
}

impl LogWriter for BoundedInMemLogStoreWriter {
    async fn init(
        &mut self,
        epoch: EpochPair,
        _pause_read_on_bootstrap: bool,
    ) -> 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<'_>> {

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Check the logs of the downstream executor/actor that owns the reader for an earlier panic or failure that dropped it.
  2. Ensure the writer and reader are created and driven by the same lifecycle (factory pairing) so neither outlives the other.
  3. Verify no code path calls writer `init()` after the streaming task has been cancelled; treat this error as terminal for the actor.
  4. Reproduce with a minimal test that drops the reader before writer init to confirm the pairing bug.

Example fix

// before: reader dropped, writer init panics/fails later
let (writer, reader) = factory.build();
drop(reader);
writer.init(epoch).await?;

// after: keep reader alive in its consumer task before writer init
let (writer, reader) = factory.build();
tokio::spawn(async move { let mut r = reader; r.init().await; /* consume */ });
writer.init(epoch).await?;
Defensive patterns

Strategy: try-catch

Validate before calling

if reader_dropped_or_terminated() { fail_actor(); } // check pairing before writer.init

Type guard

fn writer_init_ok(init_epoch_tx: &Option<oneshot::Sender<u64>>) -> bool { init_epoch_tx.is_some() }

Try / catch

match writer.init(epoch).await {
    Ok(()) => {},
    Err(e) if e.to_string().contains("unable to send init epoch") => {
        // reader side gone; fail actor and rely on recovery
        fail_actor(e);
    },
    Err(e) => return Err(e),
}

Prevention

When it happens

Trigger: 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.

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

Related errors

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